martin/linkedIn_auto_jobs_applier_with_AI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

adding pre-commit and fixing main linting errors

Browse Source
This commit is contained in:
Vincent Lordier 2024-09-02 19:38:45 +02:00
parent 165a6150f8
commit 4cad52688e
23 changed files with 409 additions and 246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -11,3 +11,4 @@ generated_cv*
.vscode
chrome_profile
answers.json
.aider*

35
.pre-commit-config.yaml Normal file
View File

@ -0,0 +1,35 @@
---
ci:
autoupdate_schedule: monthly
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.6.0 # Use the latest version or the version you prefer
hooks:
- id: trailing-whitespace
args: [--markdown-linebreak-ext=md]
- id: end-of-file-fixer
- id: check-yaml
- id: check-added-large-files
args: [--maxkb=99000]
- id: check-json
- id: check-yaml
- id: check-case-conflict
- id: detect-private-key
- id: requirements-txt-fixer
- id: check-executables-have-shebangs
- id: check-symlinks
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.6.3
hooks:
- id: ruff
args: [--fix, --exit-non-zero-on-fix, --show-fixes]
- repo: https://github.com/adrienverge/yamllint.git
rev: v1.35.1 # Use the latest version or the version you prefer
hooks:
- id: yamllint
args: [--strict]

6
.yamllint Normal file
View File

@ -0,0 +1,6 @@
---
extends: default
rules:
line-length:
max: 140

14
README.md
View File

@ -14,7 +14,7 @@
<br />
<!-- Message Clarity -->
## 🚀 Join the AIHawk Community 🚀
## 🚀 Join the AIHawk Community 🚀
Connect with like-minded individuals and get the most out of AIHawk.
@ -479,7 +479,7 @@ Using this folder as a guide can be particularly helpful for:
## Usage
0. **LinkedIn language**
To ensure the bot works, your LinkedIn language must be set to English.
2. **Data Folder:**
Ensure that your data_folder contains the following files:
- `secrets.yaml`
@ -507,15 +507,15 @@ TODO ):
## Troubleshooting
- **Carefully read logs and output :** Most of the errors are verbosely reflected just watch the output and try to find the root couse.
- **If nothing works by unknown reason:** Use tested OS. Reboot and/or update OS. Use new clean venv. Try update Python to the tested version.
- **Carefully read logs and output :** Most of the errors are verbosely reflected just watch the output and try to find the root couse.
- **If nothing works by unknown reason:** Use tested OS. Reboot and/or update OS. Use new clean venv. Try update Python to the tested version.
- **ChromeDriver Issues:** Ensure ChromeDriver is compatible with your installed Chrome version.
- **Missing Files:** Verify that all necessary files are present in the data folder.
- **Invalid YAML:** Check your YAML files for syntax errors . Try to use external YAML validators e.g. https://www.yamllint.com/
- **OpenAI endpoint isues**: Try to check possible limits\blocking at their side
- **OpenAI endpoint isues**: Try to check possible limits\blocking at their side
If you encounter any issues, you can open an issue on [GitHub](https://github.com/feder-cr/linkedIn_auto_jobs_applier_with_AI/issues).
Please add valuable details to the subject and to the description. If you need new feature then please reflect this.
Please add valuable details to the subject and to the description. If you need new feature then please reflect this.
I'll be more than happy to assist you!
## Conclusion

81
data_folder/config.yaml
View File

@ -1,42 +1,71 @@
remote: [true/false]
---
remote: true
experienceLevel:
internship: [true/false]
entry: [true/false]
associate: [true/false]
mid-senior level: [true/false]
director: [true/false]
executive: [true/false]
internship: false
entry: false
associate: false
mid-senior level: true
director: true
executive: true
jobTypes:
full-time: [true/false]
contract: [true/false]
part-time: [true/false]
temporary: [true/false]
internship: [true/false]
other: [true/false]
volunteer: [true/false]
full-time: true
contract: true
part-time: true
temporary: true
internship: false
other: true
volunteer: false
date:
all time: [true/false]
month: [true/false]
week: [true/false]
24 hours: [true/false]
all time: true
month: false
week: false
24 hours: false
positions:
- position1
- position2
- Head of Machine Learning
- Machine Learning Engineer
- Machine Learning Scientist
- Machine Learning Researcher
- Generative AI
- GenAI
- ML Engineer
- Senior ML Engineer
locations:
- Country1
- Country2
- France
- Germany
- United Kingdom
- United States
- Canada
- Australia
- New Zealand
- Singapore
- Japan
- South Korea
- China
- Switzerland
- Sweden
- Norway
- Denmark
- Finland
- Netherlands
- Belgium
- Luxembourg
- Austria
- Ireland
- Spain
- Portugal
- Italy
distance: 100
companyBlacklist:
- Company1
- Company2
- Company1
- Company2
titleBlacklist:
- word1
- word2
- word1
- word2

1
data_folder/plain_text_resume.yaml
View File

@ -1,3 +1,4 @@
---
personal_information:
name: "[Your Name]"
surname: "[Your Surname]"

3
data_folder/secrets.yaml
View File

@ -1,3 +1,4 @@
---
email: myemaillinkedin@gmail.com
password: ImpossiblePassowrd10
openai_api_key: sk-11KRr4uuTwpRGfeRTfj1T9BlbkFJjP8QTrswHU1yGruru2FR
openai_api_key: sk-11KRr4uuTwpRGfeRTfj1T9BlbkFJjP8QTrswHU1yGruru2FR

1
data_folder_example/config.yaml
View File

@ -1,3 +1,4 @@
---
remote: true
experienceLevel:

29
data_folder_example/plain_text_resume.yaml
View File

@ -1,3 +1,4 @@
---
personal_information:
name: "Liam"
surname: "Murphy"
@ -10,7 +11,7 @@ personal_information:
email: "liam.murphy@gmail.com"
github: "https://github.com/liam-murphy"
linkedin: "https://www.linkedin.com/in/liam-murphy/"
education_details:
- degree: "Bachelor's Degree"
university: "National University of Ireland, Galway"
@ -32,8 +33,12 @@ experience_details:
industry: "Blockchain Technology"
key_responsibilities:
- responsibility_1: "Co-founded and led a startup specializing in app and software development with a focus on blockchain technology"
- responsibility_2: "Provided blockchain consultations for 10+ companies, enhancing their software capabilities with secure, decentralized solutions"
- responsibility_3: "Developed blockchain applications, integrated cutting-edge technology to meet client needs and drive industry innovation"
- responsibility_2: >
Provided blockchain consultations for 10+ companies, enhancing their software capabilities
with secure, decentralized solutions
- responsibility_3: >
Developed blockchain applications, integrated cutting-edge technology to meet client needs
and drive industry innovation
skills_acquired:
- "Blockchain development"
- "Software engineering"
@ -47,7 +52,9 @@ experience_details:
key_responsibilities:
- responsibility_1: "Conducted in-depth research on IoT security, focusing on binary instrumentation and runtime monitoring"
- responsibility_2: "Performed in-depth study of the MQTT protocol and Falco"
- responsibility_3: "Developed multiple software components including MQTT packet analysis library, Falco adapter, and RML monitor in Prolog"
- responsibility_3: >
Developed multiple software components including MQTT packet analysis library, Falco adapter,
and RML monitor in Prolog
- responsibility_4: "Authored thesis 'Binary Instrumentation for Runtime Monitoring of Internet of Things Systems Using Falco'"
skills_acquired:
- "IoT security"
@ -62,8 +69,8 @@ experience_details:
industry: "Healthcare IT"
key_responsibilities:
- responsibility_1: "Integrated and enforced robust security protocols"
- responsibility_2: "Developed and maintained a critical software tool for password validation used by over 1,600 employees"
- responsibility_3: "Played an integral role in the hospital's cybersecurity team"
- responsibility_2: "Developed and maintained a critical software tool for password validation used by over 1,600 employees"
- responsibility_3: "Played an integral role in the hospital's cybersecurity team"
skills_acquired:
- "Cybersecurity"
- "Software development"
@ -71,7 +78,9 @@ experience_details:
projects:
- name: "JobBot"
description: "AI-driven tool to automate and personalize job applications on LinkedIn, gained over 3000 stars on GitHub, improving efficiency and reducing application time"
description: >
AI-driven tool to automate and personalize job applications on LinkedIn, gained over 3000 stars
on GitHub, improving efficiency and reducing application time
link: "https://github.com/liam-murphy/jobbot"
- name: "mqtt-packet-parser"
description: "Developed a Node.js module for parsing MQTT packets, improved parsing efficiency by 40%"
@ -79,7 +88,9 @@ projects:
achievements:
- name: "Winner of an Irish public competition"
description: "Won first place in a public competition with a perfect score of 70/70, securing a Software Developer position at University Hospital Galway"
description: >
Won first place in a public competition with a perfect score of 70/70, securing a Software
Developer position at University Hospital Galway
- name: "Galway Merit Scholarship"
description: "Awarded annually from 2018 to 2020 in recognition of academic excellence and contribution"
- name: "GitHub Recognition"
@ -130,4 +141,4 @@ work_preferences:
open_to_relocation: "Yes"
willing_to_complete_assessments: "Yes"
willing_to_undergo_drug_tests: "Yes"
willing_to_undergo_background_checks: "Yes"
willing_to_undergo_background_checks: "Yes"

3
data_folder_example/secrets.yaml
View File

@ -1,3 +1,4 @@
---
email: myemaillinkedin@gmail.com
password: ImpossiblePassowrd10
openai_api_key: sk-11KRr4uuTwpRGfeRTfj1T9BlbkFJjP8QTrswHU1yGruru2FR
openai_api_key: sk-11KRr4uuTwpRGfeRTfj1T9BlbkFJjP8QTrswHU1yGruru2FR

110
main.py
View File

@ -7,7 +7,7 @@ import click
from selenium import webdriver
from selenium.webdriver.chrome.service import Service as ChromeService
from webdriver_manager.chrome import ChromeDriverManager
from selenium.common.exceptions import WebDriverException, TimeoutException
from selenium.common.exceptions import WebDriverException
from lib_resume_builder_AIHawk import Resume,StyleManager,FacadeManager,ResumeGenerator
from src.utils import chromeBrowserOptions
from src.gpt import GPTAnswerer
@ -17,7 +17,7 @@ from src.linkedIn_job_manager import LinkedInJobManager
from src.job_application_profile import JobApplicationProfile
# Suppress stderr
sys.stderr = open(os.devnull, 'w')
sys.stderr = open(os.devnull, "w")
class ConfigError(Exception):
pass
@ -25,70 +25,70 @@ class ConfigError(Exception):
class ConfigValidator:
@staticmethod
def validate_email(email: str) -> bool:
return re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', email) is not None
return re.match(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$", email) is not None
@staticmethod
def validate_yaml_file(yaml_path: Path) -> dict:
try:
with open(yaml_path, 'r') as stream:
with open(yaml_path, "r") as stream:
return yaml.safe_load(stream)
except yaml.YAMLError as exc:
raise ConfigError(f"Error reading file {yaml_path}: {exc}")
except FileNotFoundError:
raise ConfigError(f"File not found: {yaml_path}")
raise ConfigError(f"Error reading file {yaml_path}: {exc}") from exc
except FileNotFoundError as exc:
raise ConfigError(f"File not found: {yaml_path}") from exc
def validate_config(config_yaml_path: Path) -> dict:
parameters = ConfigValidator.validate_yaml_file(config_yaml_path)
required_keys = {
'remote': bool,
'experienceLevel': dict,
'jobTypes': dict,
'date': dict,
'positions': list,
'locations': list,
'distance': int,
'companyBlacklist': list,
'titleBlacklist': list
"remote": bool,
"experienceLevel": dict,
"jobTypes": dict,
"date": dict,
"positions": list,
"locations": list,
"distance": int,
"companyBlacklist": list,
"titleBlacklist": list
}
for key, expected_type in required_keys.items():
if key not in parameters:
if key in ['companyBlacklist', 'titleBlacklist']:
if key in ["companyBlacklist", "titleBlacklist"]:
parameters[key] = []
else:
raise ConfigError(f"Missing or invalid key '{key}' in config file {config_yaml_path}")
elif not isinstance(parameters[key], expected_type):
if key in ['companyBlacklist', 'titleBlacklist'] and parameters[key] is None:
if key in ["companyBlacklist", "titleBlacklist"] and parameters[key] is None:
parameters[key] = []
else:
raise ConfigError(f"Invalid type for key '{key}' in config file {config_yaml_path}. Expected {expected_type}.")
experience_levels = ['internship', 'entry', 'associate', 'mid-senior level', 'director', 'executive']
experience_levels = ["internship", "entry", "associate", "mid-senior level", "director", "executive"]
for level in experience_levels:
if not isinstance(parameters['experienceLevel'].get(level), bool):
if not isinstance(parameters["experienceLevel"].get(level), bool):
raise ConfigError(f"Experience level '{level}' must be a boolean in config file {config_yaml_path}")
job_types = ['full-time', 'contract', 'part-time', 'temporary', 'internship', 'other', 'volunteer']
job_types = ["full-time", "contract", "part-time", "temporary", "internship", "other", "volunteer"]
for job_type in job_types:
if not isinstance(parameters['jobTypes'].get(job_type), bool):
if not isinstance(parameters["jobTypes"].get(job_type), bool):
raise ConfigError(f"Job type '{job_type}' must be a boolean in config file {config_yaml_path}")
date_filters = ['all time', 'month', 'week', '24 hours']
date_filters = ["all time", "month", "week", "24 hours"]
for date_filter in date_filters:
if not isinstance(parameters['date'].get(date_filter), bool):
if not isinstance(parameters["date"].get(date_filter), bool):
raise ConfigError(f"Date filter '{date_filter}' must be a boolean in config file {config_yaml_path}")
if not all(isinstance(pos, str) for pos in parameters['positions']):
if not all(isinstance(pos, str) for pos in parameters["positions"]):
raise ConfigError(f"'positions' must be a list of strings in config file {config_yaml_path}")
if not all(isinstance(loc, str) for loc in parameters['locations']):
if not all(isinstance(loc, str) for loc in parameters["locations"]):
raise ConfigError(f"'locations' must be a list of strings in config file {config_yaml_path}")
approved_distances = {0, 5, 10, 25, 50, 100}
if parameters['distance'] not in approved_distances:
if parameters["distance"] not in approved_distances:
raise ConfigError(f"Invalid distance value in config file {config_yaml_path}. Must be one of: {approved_distances}")
for blacklist in ['companyBlacklist', 'titleBlacklist']:
for blacklist in ["companyBlacklist", "titleBlacklist"]:
if not isinstance(parameters.get(blacklist), list):
raise ConfigError(f"'{blacklist}' must be a list in config file {config_yaml_path}")
if parameters[blacklist] is None:
@ -101,20 +101,20 @@ class ConfigValidator:
@staticmethod
def validate_secrets(secrets_yaml_path: Path) -> tuple:
secrets = ConfigValidator.validate_yaml_file(secrets_yaml_path)
mandatory_secrets = ['email', 'password', 'openai_api_key']
mandatory_secrets = ["email", "password", "openai_api_key"]
for secret in mandatory_secrets:
if secret not in secrets:
raise ConfigError(f"Missing secret '{secret}' in file {secrets_yaml_path}")
if not ConfigValidator.validate_email(secrets['email']):
if not ConfigValidator.validate_email(secrets["email"]):
raise ConfigError(f"Invalid email format in secrets file {secrets_yaml_path}.")
if not secrets['password']:
if not secrets["password"]:
raise ConfigError(f"Password cannot be empty in secrets file {secrets_yaml_path}.")
if not secrets['openai_api_key']:
if not secrets["openai_api_key"]:
raise ConfigError(f"OpenAI API key cannot be empty in secrets file {secrets_yaml_path}.")
return secrets['email'], str(secrets['password']), secrets['openai_api_key']
return secrets["email"], str(secrets["password"]), secrets["openai_api_key"]
class FileManager:
@staticmethod
@ -126,27 +126,27 @@ class FileManager:
if not app_data_folder.exists() or not app_data_folder.is_dir():
raise FileNotFoundError(f"Data folder not found: {app_data_folder}")
required_files = ['secrets.yaml', 'config.yaml', 'plain_text_resume.yaml']
required_files = ["secrets.yaml", "config.yaml", "plain_text_resume.yaml"]
missing_files = [file for file in required_files if not (app_data_folder / file).exists()]
if missing_files:
raise FileNotFoundError(f"Missing files in the data folder: {', '.join(missing_files)}")
output_folder = app_data_folder / 'output'
output_folder = app_data_folder / "output"
output_folder.mkdir(exist_ok=True)
return (app_data_folder / 'secrets.yaml', app_data_folder / 'config.yaml', app_data_folder / 'plain_text_resume.yaml', output_folder)
return (app_data_folder / "secrets.yaml", app_data_folder / "config.yaml", app_data_folder / "plain_text_resume.yaml", output_folder)
@staticmethod
def file_paths_to_dict(resume_file: Path | None, plain_text_resume_file: Path) -> dict:
if not plain_text_resume_file.exists():
raise FileNotFoundError(f"Plain text resume file not found: {plain_text_resume_file}")
result = {'plainTextResume': plain_text_resume_file}
result = {"plainTextResume": plain_text_resume_file}
if resume_file:
if not resume_file.exists():
raise FileNotFoundError(f"Resume file not found: {resume_file}")
result['resume'] = resume_file
result["resume"] = resume_file
return result
@ -156,22 +156,22 @@ def init_browser() -> webdriver.Chrome:
service = ChromeService(ChromeDriverManager().install())
return webdriver.Chrome(service=service, options=options)
except Exception as e:
raise RuntimeError(f"Failed to initialize browser: {str(e)}")
raise RuntimeError(f"Failed to initialize browser: {str(e)}") from e
def create_and_run_bot(email: str, password: str, parameters: dict, openai_api_key: str):
try:
style_manager = StyleManager()
resume_generator = ResumeGenerator()
with open(parameters['uploads']['plainTextResume'], "r") as file:
with open(parameters["uploads"]["plainTextResume"], "r") as file:
plain_text_resume = file.read()
resume_object = Resume(plain_text_resume)
resume_generator_manager = FacadeManager(openai_api_key, style_manager, resume_generator, resume_object, Path("data_folder/output"))
os.system('cls' if os.name == 'nt' else 'clear')
os.system("cls" if os.name == "nt" else "clear")
resume_generator_manager.choose_style()
os.system('cls' if os.name == 'nt' else 'clear')
os.system("cls" if os.name == "nt" else "clear")
job_application_profile_object = JobApplicationProfile(plain_text_resume)
browser = init_browser()
login_component = LinkedInAuthenticator(browser)
apply_component = LinkedInJobManager(browser)
@ -186,22 +186,22 @@ def create_and_run_bot(email: str, password: str, parameters: dict, openai_api_k
except WebDriverException as e:
print(f"WebDriver error occurred: {e}")
except Exception as e:
raise RuntimeError(f"Error running the bot: {str(e)}")
raise RuntimeError(f"Error running the bot: {str(e)}") from e
@click.command()
@click.option('--resume', type=click.Path(exists=True, file_okay=True, dir_okay=False, path_type=Path), help="Path to the resume PDF file")
@click.option("--resume", type=click.Path(exists=True, file_okay=True, dir_okay=False, path_type=Path), help="Path to the resume PDF file")
def main(resume: Path = None):
try:
data_folder = Path("data_folder")
secrets_file, config_file, plain_text_resume_file, output_folder = FileManager.validate_data_folder(data_folder)
parameters = ConfigValidator.validate_config(config_file)
email, password, openai_api_key = ConfigValidator.validate_secrets(secrets_file)
parameters['uploads'] = FileManager.file_paths_to_dict(resume, plain_text_resume_file)
parameters['outputFileDirectory'] = output_folder
parameters["uploads"] = FileManager.file_paths_to_dict(resume, plain_text_resume_file)
parameters["outputFileDirectory"] = output_folder
create_and_run_bot(email, password, parameters, openai_api_key)
except ConfigError as ce:
print(f"Configuration error: {str(ce)}")

2
pyproject.toml Normal file
View File

@ -0,0 +1,2 @@
[tool.ruff]
line-length = 140

BIN
requirements.txt
View File

Binary file not shown.

78
ruff.toml Normal file
View File

@ -0,0 +1,78 @@
# Exclude a variety of commonly ignored directories.
exclude = [
".bzr",
".direnv",
".eggs",
".git",
".git-rewrite",
".hg",
".ipynb_checkpoints",
".mypy_cache",
".nox",
".pants.d",
".pyenv",
".pytest_cache",
".pytype",
".ruff_cache",
".svn",
".tox",
".venv",
".vscode",
"__pypackages__",
"_build",
"buck-out",
"build",
"dist",
"node_modules",
"site-packages",
"venv"
]
indent-width = 4
# Same as Black.
line-length = 140
# Assume Python 3.8
target-version = "py311"
[format]
# Enable auto-formatting of code examples in docstrings. Markdown,
# reStructuredText code/literal blocks and doctests are all supported.
#
# This is currently disabled by default, but it is planned for this
# to be opt-out in the future.
# Set the line length limit used when formatting code snippets in
# docstrings.
#
# This only has an effect when the `docstring-code-format` setting is
# enabled.
# docstring-code-line-length = "dynamic"
# Like Black, indent with spaces, rather than tabs.
indent-style = "space"
# Like Black, automatically detect the appropriate line ending.
line-ending = "auto"
# Like Black, use double quotes for strings.
quote-style = "double"
# Like Black, respect magic trailing commas.
skip-magic-trailing-comma = false
[lint]
# Allow unused variables when underscore-prefixed.
dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
# Allow fix for all enabled rules (when `--fix`) is provided.
fixable = ["ALL"]
# 2. Avoid enforcing line-length violations (`E501`)
ignore = ["E501"]
# 1. Enable flake8-bugbear (`B`) rules, in addition to the defaults.
select = ["E4", "E7", "E9", "F", "B", "Q"]
unfixable = []
[lint.flake8-quotes]
docstring-quotes = "double"
[lint.isort]
case-sensitive = true
# 4. Ignore `E402` (import violations) in all `__init__.py` files, and in select subdirectories.
[lint.per-file-ignores]
"**/{tests,docs,tools}/*" = ["E402"]
"__init__.py" = ["E402"]

14
src/gpt.py
View File

@ -19,7 +19,7 @@ load_dotenv()
class LLMLogger:
def __init__(self, llm: ChatOpenAI):
self.llm = llm
@ -148,7 +148,7 @@ class GPTAnswerer:
def set_job_application_profile(self, job_application_profile):
self.job_application_profile = job_application_profile
def summarize_job_description(self, text: str) -> str:
strings.summarize_prompt_template = self._preprocess_template_string(
strings.summarize_prompt_template
@ -157,11 +157,11 @@ class GPTAnswerer:
chain = prompt | self.llm_cheap | StrOutputParser()
output = chain.invoke({"text": text})
return output
def _create_chain(self, template: str):
prompt = ChatPromptTemplate.from_template(template)
return prompt | self.llm_cheap | StrOutputParser()
def answer_question_textual_wide_range(self, question: str) -> str:
# Define chains for each section of the resume
chains = {
@ -182,7 +182,7 @@ class GPTAnswerer:
section_prompt = """
You are assisting a bot designed to automatically apply for jobs on LinkedIn. The bot receives various questions about job applications and needs to determine the most relevant section of the resume to provide an accurate response.
For the following question: '{question}', determine which section of the resume is most relevant.
For the following question: '{question}', determine which section of the resume is most relevant.
Respond with exactly one of the following options:
- Personal information
- Self Identification
@ -308,12 +308,12 @@ class GPTAnswerer:
output_str = chain.invoke({"resume": self.resume, "question": question, "options": options})
best_option = self.find_best_match(output_str, options)
return best_option
def resume_or_cover(self, phrase: str) -> str:
# Define the prompt template
prompt_template = """
Given the following phrase, respond with only 'resume' if the phrase is about a resume, or 'cover' if it's about a cover letter. Do not provide any additional information or explanations.
phrase: {phrase}
"""
prompt = ChatPromptTemplate.from_template(prompt_template)

4
src/job.py
View File

@ -27,12 +27,12 @@ class Job:
"""
job_information = f"""
# Job Description
## Job Information
## Job Information
- Position: {self.title}
- At: {self.company}
- Location: {self.location}
- Recruiter Profile: {self.recruiter_link or 'Not available'}
## Description
{self.description or 'No description provided.'}
"""

11
src/job_application_profile.py
View File

@ -1,5 +1,4 @@
from dataclasses import dataclass
from typing import Dict, List
import yaml
@dataclass
@ -59,7 +58,7 @@ class JobApplicationProfile:
# Process self_identification
try:
self.self_identification = SelfIdentification(**data['self_identification'])
self.self_identification = SelfIdentification(**data["self_identification"])
except KeyError as e:
raise KeyError(f"Required field {e} is missing in self_identification data.") from e
except TypeError as e:
@ -71,7 +70,7 @@ class JobApplicationProfile:
# Process legal_authorization
try:
self.legal_authorization = LegalAuthorization(**data['legal_authorization'])
self.legal_authorization = LegalAuthorization(**data["legal_authorization"])
except KeyError as e:
raise KeyError(f"Required field {e} is missing in legal_authorization data.") from e
except TypeError as e:
@ -83,7 +82,7 @@ class JobApplicationProfile:
# Process work_preferences
try:
self.work_preferences = WorkPreferences(**data['work_preferences'])
self.work_preferences = WorkPreferences(**data["work_preferences"])
except KeyError as e:
raise KeyError(f"Required field {e} is missing in work_preferences data.") from e
except TypeError as e:
@ -95,7 +94,7 @@ class JobApplicationProfile:
# Process availability
try:
self.availability = Availability(**data['availability'])
self.availability = Availability(**data["availability"])
except KeyError as e:
raise KeyError(f"Required field {e} is missing in availability data.") from e
except TypeError as e:
@ -107,7 +106,7 @@ class JobApplicationProfile:
# Process salary_expectations
try:
self.salary_expectations = SalaryExpectations(**data['salary_expectations'])
self.salary_expectations = SalaryExpectations(**data["salary_expectations"])
except KeyError as e:
raise KeyError(f"Required field {e} is missing in salary_expectations data.") from e
except TypeError as e:

18
src/linkedIn_authenticator.py
View File

@ -5,7 +5,7 @@ from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
class LinkedInAuthenticator:
def __init__(self, driver=None):
self.driver = driver
self.email = ""
@ -17,7 +17,7 @@ class LinkedInAuthenticator:
def start(self):
print("Starting Chrome browser to log in to LinkedIn.")
self.driver.get('https://www.linkedin.com')
self.driver.get("https://www.linkedin.com")
self.wait_for_page_load()
if not self.is_logged_in():
self.handle_login()
@ -54,24 +54,24 @@ class LinkedInAuthenticator:
def handle_security_check(self):
try:
WebDriverWait(self.driver, 10).until(
EC.url_contains('https://www.linkedin.com/checkpoint/challengesV2/')
EC.url_contains("https://www.linkedin.com/checkpoint/challengesV2/")
)
print("Security checkpoint detected. Please complete the challenge.")
WebDriverWait(self.driver, 300).until(
EC.url_contains('https://www.linkedin.com/feed/')
EC.url_contains("https://www.linkedin.com/feed/")
)
print("Security check completed")
except TimeoutException:
print("Security check not completed. Please try again later.")
def is_logged_in(self):
self.driver.get('https://www.linkedin.com/feed')
self.driver.get("https://www.linkedin.com/feed")
try:
WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.CLASS_NAME, 'share-box-feed-entry__trigger'))
EC.presence_of_element_located((By.CLASS_NAME, "share-box-feed-entry__trigger"))
)
buttons = self.driver.find_elements(By.CLASS_NAME, 'share-box-feed-entry__trigger')
if any(button.text.strip() == 'Start a post' for button in buttons):
buttons = self.driver.find_elements(By.CLASS_NAME, "share-box-feed-entry__trigger")
if any(button.text.strip() == "Start a post" for button in buttons):
print("User is already logged in.")
return True
except TimeoutException:
@ -81,7 +81,7 @@ class LinkedInAuthenticator:
def wait_for_page_load(self, timeout=10):
try:
WebDriverWait(self.driver, timeout).until(
lambda d: d.execute_script('return document.readyState') == 'complete'
lambda d: d.execute_script("return document.readyState") == "complete"
)
except TimeoutException:
print("Page load timed out.")

4
src/linkedIn_bot_facade.py
View File

@ -55,13 +55,13 @@ class LinkedInBotFacade:
self.state.parameters_set = True
def start_login(self):
self.state.validate_state(['credentials_set'])
self.state.validate_state(["credentials_set"])
self.login_component.set_secrets(self.email, self.password)
self.login_component.start()
self.state.logged_in = True
def start_apply(self):
self.state.validate_state(['logged_in', 'job_application_profile_set', 'gpt_answerer_set', 'parameters_set'])
self.state.validate_state(["logged_in", "job_application_profile_set", "gpt_answerer_set", "parameters_set"])
self.apply_component.start_applying()
def _validate_non_empty(self, value, name):

152
src/linkedIn_easy_applier.py
View File

@ -6,13 +6,11 @@ import re
import tempfile
import time
import traceback
from datetime import date
from typing import List, Optional, Any, Tuple
from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.remote.webelement import WebElement
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import Select, WebDriverWait
@ -32,10 +30,10 @@ class LinkedInEasyApplier:
def _load_questions_from_json(self) -> List[dict]:
output_file = 'answers.json'
output_file = "answers.json"
try:
try:
with open(output_file, 'r') as f:
with open(output_file, "r") as f:
try:
data = json.load(f)
if not isinstance(data, list):
@ -45,9 +43,9 @@ class LinkedInEasyApplier:
except FileNotFoundError:
data = []
return data
except Exception:
except Exception as e:
tb_str = traceback.format_exc()
raise Exception(f"Error loading questions data from JSON file: \nTraceback:\n{tb_str}")
raise Exception(f"Error loading questions data from JSON file: \nTraceback:\n{tb_str}") from e
def job_apply(self, job: Any):
@ -61,10 +59,10 @@ class LinkedInEasyApplier:
actions.move_to_element(easy_apply_button).click().perform()
self.gpt_answerer.set_job(job)
self._fill_application_form(job)
except Exception:
except Exception as e:
tb_str = traceback.format_exc()
self._discard_application()
raise Exception(f"Failed to apply to job! Original exception: \nTraceback:\n{tb_str}")
raise Exception(f"Failed to apply to job! Original exception: \nTraceback:\n{tb_str}") from e
def _find_easy_apply_button(self) -> WebElement:
attempt = 0
@ -83,14 +81,14 @@ class LinkedInEasyApplier:
)
)
return button
except Exception as e:
except Exception:
pass
if attempt == 0:
self.driver.refresh()
time.sleep(3)
time.sleep(3)
attempt += 1
raise Exception("No clickable 'Easy Apply' button found")
def _get_job_description(self) -> str:
try:
@ -98,14 +96,14 @@ class LinkedInEasyApplier:
actions = ActionChains(self.driver)
actions.move_to_element(see_more_button).click().perform()
time.sleep(2)
description = self.driver.find_element(By.CLASS_NAME, 'jobs-description-content__text').text
description = self.driver.find_element(By.CLASS_NAME, "jobs-description-content__text").text
return description
except NoSuchElementException:
except NoSuchElementException as e:
tb_str = traceback.format_exc()
raise Exception("Job description 'See more' button not found: \nTraceback:\n{tb_str}")
except Exception:
raise Exception("Job description 'See more' button not found: \nTraceback:\n{tb_str}") from e
except Exception as e:
tb_str = traceback.format_exc()
raise Exception(f"Error getting Job description: \nTraceback:\n{tb_str}")
raise Exception(f"Error getting Job description: \nTraceback:\n{tb_str}") from e
def _get_job_recruiter(self):
@ -114,13 +112,13 @@ class LinkedInEasyApplier:
EC.presence_of_element_located((By.XPATH, '//h2[text()="Meet the hiring team"]'))
)
recruiter_element = hiring_team_section.find_element(By.XPATH, './/following::a[contains(@href, "linkedin.com/in/")]')
recruiter_link = recruiter_element.get_attribute('href')
recruiter_link = recruiter_element.get_attribute("href")
return recruiter_link
except Exception as e:
except Exception:
return ""
def _scroll_page(self) -> None:
scrollable_element = self.driver.find_element(By.TAG_NAME, 'html')
scrollable_element = self.driver.find_element(By.TAG_NAME, "html")
utils.scroll_slow(self.driver, scrollable_element, step=300, reverse=False)
utils.scroll_slow(self.driver, scrollable_element, step=300, reverse=True)
@ -133,7 +131,7 @@ class LinkedInEasyApplier:
def _next_or_submit(self):
next_button = self.driver.find_element(By.CLASS_NAME, "artdeco-button--primary")
button_text = next_button.text.lower()
if 'submit application' in button_text:
if "submit application" in button_text:
self._unfollow_company()
time.sleep(random.uniform(1.5, 2.5))
next_button.click()
@ -149,29 +147,29 @@ class LinkedInEasyApplier:
follow_checkbox = self.driver.find_element(
By.XPATH, "//label[contains(.,'to stay up to date with their page.')]")
follow_checkbox.click()
except Exception as e:
except Exception:
pass
def _check_for_errors(self) -> None:
error_elements = self.driver.find_elements(By.CLASS_NAME, 'artdeco-inline-feedback--error')
error_elements = self.driver.find_elements(By.CLASS_NAME, "artdeco-inline-feedback--error")
if error_elements:
raise Exception(f"Failed answering or file upload. {str([e.text for e in error_elements])}")
def _discard_application(self) -> None:
try:
self.driver.find_element(By.CLASS_NAME, 'artdeco-modal__dismiss').click()
self.driver.find_element(By.CLASS_NAME, "artdeco-modal__dismiss").click()
time.sleep(random.uniform(3, 5))
self.driver.find_elements(By.CLASS_NAME, 'artdeco-modal__confirm-dialog-btn')[0].click()
self.driver.find_elements(By.CLASS_NAME, "artdeco-modal__confirm-dialog-btn")[0].click()
time.sleep(random.uniform(3, 5))
except Exception as e:
except Exception:
pass
def fill_up(self, job) -> None:
easy_apply_content = self.driver.find_element(By.CLASS_NAME, 'jobs-easy-apply-content')
pb4_elements = easy_apply_content.find_elements(By.CLASS_NAME, 'pb4')
easy_apply_content = self.driver.find_element(By.CLASS_NAME, "jobs-easy-apply-content")
pb4_elements = easy_apply_content.find_elements(By.CLASS_NAME, "pb4")
for element in pb4_elements:
self._process_form_element(element, job)
def _process_form_element(self, element: WebElement, job) -> None:
if self._is_upload_field(element):
self._handle_upload_fields(element, job)
@ -187,16 +185,16 @@ class LinkedInEasyApplier:
parent = element.find_element(By.XPATH, "..")
self.driver.execute_script("arguments[0].classList.remove('hidden')", element)
output = self.gpt_answerer.resume_or_cover(parent.text.lower())
if 'resume' in output:
if "resume" in output:
if self.resume_path is not None and self.resume_path.resolve().is_file():
element.send_keys(str(self.resume_path.resolve()))
else:
self._create_and_upload_resume(element, job)
elif 'cover' in output:
elif "cover" in output:
self._create_and_upload_cover_letter(element)
def _create_and_upload_resume(self, element, job):
folder_path = 'generated_cv'
folder_path = "generated_cv"
os.makedirs(folder_path, exist_ok=True)
try:
file_path_pdf = os.path.join(folder_path, f"CV_{random.randint(0, 9999)}.pdf")
@ -205,13 +203,13 @@ class LinkedInEasyApplier:
element.send_keys(os.path.abspath(file_path_pdf))
job.pdf_path = os.path.abspath(file_path_pdf)
time.sleep(2)
except Exception:
except Exception as e:
tb_str = traceback.format_exc()
raise Exception(f"Upload failed: \nTraceback:\n{tb_str}")
raise Exception(f"Upload failed: \nTraceback:\n{tb_str}") from e
def _create_and_upload_cover_letter(self, element: WebElement) -> None:
cover_letter = self.gpt_answerer.answer_question_textual_wide_range("Write a cover letter")
with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as temp_pdf_file:
with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as temp_pdf_file:
letter_path = temp_pdf_file.name
c = canvas.Canvas(letter_path, pagesize=letter)
_, height = letter
@ -223,10 +221,10 @@ class LinkedInEasyApplier:
element.send_keys(letter_path)
def _fill_additional_questions(self) -> None:
form_sections = self.driver.find_elements(By.CLASS_NAME, 'jobs-easy-apply-form-section__grouping')
form_sections = self.driver.find_elements(By.CLASS_NAME, "jobs-easy-apply-form-section__grouping")
for section in form_sections:
self._process_form_section(section)
def _process_form_section(self, section: WebElement) -> None:
if self._handle_terms_of_service(section):
@ -241,61 +239,61 @@ class LinkedInEasyApplier:
return
def _handle_terms_of_service(self, element: WebElement) -> bool:
checkbox = element.find_elements(By.TAG_NAME, 'label')
if checkbox and any(term in checkbox[0].text.lower() for term in ['terms of service', 'privacy policy', 'terms of use']):
checkbox = element.find_elements(By.TAG_NAME, "label")
if checkbox and any(term in checkbox[0].text.lower() for term in ["terms of service", "privacy policy", "terms of use"]):
checkbox[0].click()
return True
return False
def _find_and_handle_radio_question(self, section: WebElement) -> bool:
question = section.find_element(By.CLASS_NAME, 'jobs-easy-apply-form-element')
radios = question.find_elements(By.CLASS_NAME, 'fb-text-selectable__option')
question = section.find_element(By.CLASS_NAME, "jobs-easy-apply-form-element")
radios = question.find_elements(By.CLASS_NAME, "fb-text-selectable__option")
if radios:
question_text = section.text.lower()
options = [radio.text.lower() for radio in radios]
existing_answer = None
for item in self.all_data:
if self._sanitize_text(question_text) in item['question'] and item['type'] == 'radio':
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'])
self._select_radio(radios, existing_answer["answer"])
return True
answer = self.gpt_answerer.answer_question_from_options(question_text, options)
self._save_questions_to_json({'type': 'radio', 'question': question_text, 'answer': answer})
self._save_questions_to_json({"type": "radio", "question": question_text, "answer": answer})
self._select_radio(radios, answer)
return True
return False
def _find_and_handle_textbox_question(self, section: WebElement) -> bool:
text_fields = section.find_elements(By.TAG_NAME, 'input') + section.find_elements(By.TAG_NAME, 'textarea')
text_fields = section.find_elements(By.TAG_NAME, "input") + section.find_elements(By.TAG_NAME, "textarea")
if text_fields:
text_field = text_fields[0]
question_text = section.find_element(By.TAG_NAME, 'label').text.lower()
question_text = section.find_element(By.TAG_NAME, "label").text.lower()
is_numeric = self._is_numeric_field(text_field)
if is_numeric:
question_type = 'numeric'
question_type = "numeric"
answer = self.gpt_answerer.answer_question_numeric(question_text)
else:
question_type = 'textbox'
question_type = "textbox"
answer = self.gpt_answerer.answer_question_textual_wide_range(question_text)
existing_answer = None
for item in self.all_data:
if item['question'] == self._sanitize_text(question_text) and item['type'] == question_type:
if item["question"] == self._sanitize_text(question_text) and item["type"] == question_type:
existing_answer = item
break
if existing_answer:
self._enter_text(text_field, existing_answer['answer'])
self._enter_text(text_field, existing_answer["answer"])
return True
self._save_questions_to_json({'type': question_type, 'question': question_text, 'answer': answer})
self._save_questions_to_json({"type": question_type, "question": question_text, "answer": answer})
self._enter_text(text_field, answer)
return True
return False
def _find_and_handle_date_question(self, section: WebElement) -> bool:
date_fields = section.find_elements(By.CLASS_NAME, 'artdeco-datepicker__input ')
date_fields = section.find_elements(By.CLASS_NAME, "artdeco-datepicker__input ")
if date_fields:
date_field = date_fields[0]
question_text = section.text.lower()
@ -305,49 +303,49 @@ class LinkedInEasyApplier:
existing_answer = None
for item in self.all_data:
if self._sanitize_text(question_text) in item['question'] and item['type'] == 'date':
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'])
self._enter_text(date_field, existing_answer["answer"])
return True
self._save_questions_to_json({'type': 'date', 'question': question_text, 'answer': answer_text})
self._save_questions_to_json({"type": "date", "question": question_text, "answer": answer_text})
self._enter_text(date_field, answer_text)
return True
return False
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()
dropdown = question.find_element(By.TAG_NAME, 'select')
question = section.find_element(By.CLASS_NAME, "jobs-easy-apply-form-element")
question_text = question.find_element(By.TAG_NAME, "label").text.lower()
dropdown = question.find_element(By.TAG_NAME, "select")
if dropdown:
select = Select(dropdown)
options = [option.text for option in select.options]
existing_answer = None
for item in self.all_data:
if self._sanitize_text(question_text) in item['question'] and item['type'] == 'dropdown':
if self._sanitize_text(question_text) in item["question"] and item["type"] == "dropdown":
existing_answer = item
break
if existing_answer:
self._select_dropdown_option(dropdown, existing_answer['answer'])
self._select_dropdown_option(dropdown, existing_answer["answer"])
return True
answer = self.gpt_answerer.answer_question_from_options(question_text, options)
self._save_questions_to_json({'type': 'dropdown', 'question': question_text, 'answer': answer})
self._save_questions_to_json({"type": "dropdown", "question": question_text, "answer": answer})
self._select_dropdown_option(dropdown, answer)
return True
except Exception:
return False
def _is_numeric_field(self, field: WebElement) -> bool:
field_type = field.get_attribute('type').lower()
if 'numeric' in field_type:
field_type = field.get_attribute("type").lower()
if "numeric" in field_type:
return True
class_attribute = field.get_attribute("id")
return class_attribute and 'numeric' in class_attribute
return class_attribute and "numeric" in class_attribute
def _enter_text(self, element: WebElement, text: str) -> None:
element.clear()
@ -356,20 +354,20 @@ class LinkedInEasyApplier:
def _select_radio(self, radios: List[WebElement], answer: str) -> None:
for radio in radios:
if answer in radio.text.lower():
radio.find_element(By.TAG_NAME, 'label').click()
radio.find_element(By.TAG_NAME, "label").click()
return
radios[-1].find_element(By.TAG_NAME, 'label').click()
radios[-1].find_element(By.TAG_NAME, "label").click()
def _select_dropdown_option(self, element: WebElement, text: str) -> None:
select = Select(element)
select.select_by_visible_text(text)
def _save_questions_to_json(self, question_data: dict) -> None:
output_file = 'answers.json'
question_data['question'] = self._sanitize_text(question_data['question'])
output_file = "answers.json"
question_data["question"] = self._sanitize_text(question_data["question"])
try:
try:
with open(output_file, 'r') as f:
with open(output_file, "r") as f:
try:
data = json.load(f)
if not isinstance(data, list):
@ -379,19 +377,19 @@ class LinkedInEasyApplier:
except FileNotFoundError:
data = []
data.append(question_data)
with open(output_file, 'w') as f:
with open(output_file, "w") as f:
json.dump(data, f, indent=4)
except Exception:
except Exception as e:
tb_str = traceback.format_exc()
raise Exception(f"Error saving questions data to JSON file: \nTraceback:\n{tb_str}")
raise Exception(f"Error saving questions data to JSON file: \nTraceback:\n{tb_str}") from e
def _sanitize_text(self, text: str) -> str:
sanitized_text = text.lower()
sanitized_text = sanitized_text.strip()
sanitized_text = sanitized_text.replace('"', '')
sanitized_text = sanitized_text.replace('\\', '')
sanitized_text = re.sub(r'[\x00-\x1F\x7F]', '', sanitized_text)
sanitized_text = sanitized_text.replace('\n', ' ').replace('\r', '')
sanitized_text = sanitized_text.rstrip(',')
sanitized_text = sanitized_text.replace('"', "")
sanitized_text = sanitized_text.replace("\\", "")
sanitized_text = re.sub(r"[\x00-\x1F\x7F]", "", sanitized_text)
sanitized_text = sanitized_text.replace("\n", " ").replace("\r", "")
sanitized_text = sanitized_text.rstrip(",")
return sanitized_text

62
src/linkedIn_job_manager.py
View File

@ -32,18 +32,18 @@ class LinkedInJobManager:
self.easy_applier_component = None
def set_parameters(self, parameters):
self.company_blacklist = parameters.get('companyBlacklist', []) or []
self.title_blacklist = parameters.get('titleBlacklist', []) or []
self.positions = parameters.get('positions', [])
self.locations = parameters.get('locations', [])
self.company_blacklist = parameters.get("companyBlacklist", []) or []
self.title_blacklist = parameters.get("titleBlacklist", []) or []
self.positions = parameters.get("positions", [])
self.locations = parameters.get("locations", [])
self.base_search_url = self.get_base_search_url(parameters)
self.seen_jobs = []
resume_path = parameters.get('uploads', {}).get('resume', None)
resume_path = parameters.get("uploads", {}).get("resume", None)
if resume_path is not None and Path(resume_path).exists():
self.resume_path = Path(resume_path)
else:
self.resume_path = None
self.output_file_directory = Path(parameters['outputFileDirectory'])
self.output_file_directory = Path(parameters["outputFileDirectory"])
self.env_config = EnvironmentKeys()
#self.old_question()
@ -115,19 +115,19 @@ class LinkedInJobManager:
def apply_jobs(self):
try:
no_jobs_element = self.driver.find_element(By.CLASS_NAME, 'jobs-search-two-pane__no-results-banner--expand')
if 'No matching jobs found' in no_jobs_element.text or 'unfortunately, things aren' in self.driver.page_source.lower():
no_jobs_element = self.driver.find_element(By.CLASS_NAME, "jobs-search-two-pane__no-results-banner--expand")
if "No matching jobs found" in no_jobs_element.text or "unfortunately, things aren" in self.driver.page_source.lower():
raise Exception("No more jobs on this page")
except NoSuchElementException:
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)
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')
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:
raise Exception("No job class elements found on page")
job_list = [Job(*self.extract_job_information_from_tile(job_element)) for job_element in job_list_elements]
job_list = [Job(*self.extract_job_information_from_tile(job_element)) for job_element in job_list_elements]
for job in job_list:
if self.is_blacklisted(job.title, job.company, job.link):
utils.printyellow(f"Blacklisted {job.title} at {job.company}, skipping...")
@ -137,11 +137,11 @@ class LinkedInJobManager:
if job.apply_method not in {"Continue", "Applied", "Apply"}:
self.easy_applier_component.job_apply(job)
self.write_to_file(job, "success")
except Exception as e:
except Exception:
utils.printred(traceback.format_exc())
self.write_to_file(job, "failed")
continue
def write_to_file(self, job, file_name):
pdf_path = Path(job.pdf_path).resolve()
pdf_path = pdf_path.as_uri()
@ -155,10 +155,10 @@ class LinkedInJobManager:
}
file_path = self.output_file_directory / f"{file_name}.json"
if not file_path.exists():
with open(file_path, 'w', encoding='utf-8') as f:
with open(file_path, "w", encoding="utf-8") as f:
json.dump([data], f, indent=4)
else:
with open(file_path, 'r+', encoding='utf-8') as f:
with open(file_path, "r+", encoding="utf-8") as f:
try:
existing_data = json.load(f)
except json.JSONDecodeError:
@ -170,13 +170,13 @@ class LinkedInJobManager:
def get_base_search_url(self, parameters):
url_parts = []
if parameters['remote']:
if parameters["remote"]:
url_parts.append("f_CF=f_WRA")
experience_levels = [str(i+1) for i, (level, v) in enumerate(parameters.get('experienceLevel', {}).items()) if v]
experience_levels = [str(i+1) for i, (level, v) in enumerate(parameters.get("experienceLevel", {}).items()) if v]
if experience_levels:
url_parts.append(f"f_E={','.join(experience_levels)}")
url_parts.append(f"distance={parameters['distance']}")
job_types = [key[0].upper() for key, value in parameters.get('jobTypes', {}).items() if value]
job_types = [key[0].upper() for key, value in parameters.get("jobTypes", {}).items() if value]
if job_types:
url_parts.append(f"f_JT={','.join(job_types)}")
date_mapping = {
@ -185,35 +185,35 @@ class LinkedInJobManager:
"week": "&f_TPR=r604800",
"24 hours": "&f_TPR=r86400"
}
date_param = next((v for k, v in date_mapping.items() if parameters.get('date', {}).get(k)), "")
date_param = next((v for k, v in date_mapping.items() if parameters.get("date", {}).get(k)), "")
url_parts.append("f_LF=f_AL") # Easy Apply
base_url = "&".join(url_parts)
return f"?{base_url}{date_param}"
def next_job_page(self, position, location, job_page):
self.driver.get(f"https://www.linkedin.com/jobs/search/{self.base_search_url}&keywords={position}{location}&start={job_page * 25}")
def extract_job_information_from_tile(self, job_tile):
job_title, company, job_location, apply_method, link = "", "", "", "", ""
try:
job_title = job_tile.find_element(By.CLASS_NAME, 'job-card-list__title').text
link = job_tile.find_element(By.CLASS_NAME, 'job-card-list__title').get_attribute('href').split('?')[0]
company = job_tile.find_element(By.CLASS_NAME, 'job-card-container__primary-description').text
except:
job_title = job_tile.find_element(By.CLASS_NAME, "job-card-list__title").text
link = job_tile.find_element(By.CLASS_NAME, "job-card-list__title").get_attribute("href").split("?")[0]
company = job_tile.find_element(By.CLASS_NAME, "job-card-container__primary-description").text
except NoSuchElementException:
pass
try:
job_location = job_tile.find_element(By.CLASS_NAME, 'job-card-container__metadata-item').text
except:
job_location = job_tile.find_element(By.CLASS_NAME, "job-card-container__metadata-item").text
except NoSuchElementException:
pass
try:
apply_method = job_tile.find_element(By.CLASS_NAME, 'job-card-container__apply-method').text
except:
apply_method = job_tile.find_element(By.CLASS_NAME, "job-card-container__apply-method").text
except NoSuchElementException:
apply_method = "Applied"
return job_title, company, job_location, link, apply_method
def is_blacklisted(self, job_title, company, link):
job_title_words = job_title.lower().split(' ')
job_title_words = job_title.lower().split(" ")
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

18
src/strings.py
View File

@ -301,7 +301,7 @@ How many years of experience do you have with IoT?
```
## Curriculum
I had a degree in computer science.
I had a degree in computer science.
## Question
@ -333,7 +333,7 @@ How many years of experience do you have with AI?
{resume_jobs}
{resume_projects}
```
## Question:
{question}
@ -386,30 +386,30 @@ The objective is to fix the text of a form input on a web page.
{question}
## Input
{input}
{input}
## Error
{error}
{error}
## Fixed Input
"""
func_summarize_prompt_template = """
Following are two texts, one with placeholders and one without, the second text uses information from the first text to fill the placeholders.
## Rules
- A placeholder is a string like "[[placeholder]]". E.g. "[[company]]", "[[job_title]]", "[[years_of_experience]]"...
- The task is to remove the placeholders from the text.
- If there is no information to fill a placeholder, remove the placeholder, and adapt the text accordingly.
- No placeholders should remain in the text.
## Example
Text with placeholders: "I'm a software engineer engineer with 10 years of experience on [placeholder] and [placeholder]."
Text without placeholders: "I'm a software engineer with 10 years of experience."
-----
## Text with placeholders:
{text_with_placeholders}
## Text without placeholders:"""

8
src/utils.py
View File

@ -33,7 +33,7 @@ def scroll_slow(driver, scrollable_element, start=0, end=3600, step=100, reverse
return
if (step > 0 and start >= end) or (step < 0 and start <= end):
print("No scrolling will occur due to incorrect start/end values.")
return
return
for position in range(start, end, step):
try:
driver.execute_script(script_scroll_to, scrollable_element, position)
@ -67,7 +67,7 @@ def chromeBrowserOptions():
options.add_argument("--disable-autofill") # Disabilita l'autocompletamento dei moduli
options.add_argument("--disable-plugins") # Disabilita i plugin del browser
options.add_argument("--disable-animations") # Disabilita le animazioni
options.add_argument("--disable-cache") # Disabilita la cache
options.add_argument("--disable-cache") # Disabilita la cache
options.add_experimental_option("excludeSwitches", ["enable-automation", "enable-logging"]) # Esclude switch della modalità automatica e logging
# Preferenze per contenuti
@ -80,7 +80,7 @@ def chromeBrowserOptions():
if len(chromeProfilePath) > 0:
initialPath = os.path.dirname(chromeProfilePath)
profileDir = os.path.basename(chromeProfilePath)
options.add_argument('--user-data-dir=' + initialPath)
options.add_argument("--user-data-dir=" + initialPath)
options.add_argument("--profile-directory=" + profileDir)
else:
options.add_argument("--incognito")
@ -100,4 +100,4 @@ def printyellow(text):
YELLOW = "\033[93m"
RESET = "\033[0m"
# Stampa il testo in giallo
print(f"{YELLOW}{text}{RESET}")
print(f"{YELLOW}{text}{RESET}")