forked from AIHawk-FOSS/Auto_Jobs_Applier_AI_Agent
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
223 lines (182 loc) · 10.5 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
import os
import re
import sys
from pathlib import Path
import yaml
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 lib_resume_builder_AIHawk import Resume,StyleManager,FacadeManager,ResumeGenerator
from src.utils import chromeBrowserOptions
from src.gpt import GPTAnswerer
from src.linkedIn_authenticator import LinkedInAuthenticator
from src.linkedIn_bot_facade import LinkedInBotFacade
from src.linkedIn_job_manager import LinkedInJobManager
from src.job_application_profile import JobApplicationProfile
# Suppress stderr
sys.stderr = open(os.devnull, 'w')
class ConfigError(Exception):
pass
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
@staticmethod
def validate_yaml_file(yaml_path: Path) -> dict:
try:
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}")
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
}
for key, expected_type in required_keys.items():
if key not in parameters:
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:
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']
for level in experience_levels:
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']
for job_type in job_types:
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']
for date_filter in date_filters:
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']):
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']):
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:
raise ConfigError(f"Invalid distance value in config file {config_yaml_path}. Must be one of: {approved_distances}")
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:
parameters[blacklist] = []
return parameters
@staticmethod
def validate_secrets(secrets_yaml_path: Path) -> tuple:
secrets = ConfigValidator.validate_yaml_file(secrets_yaml_path)
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']):
raise ConfigError(f"Invalid email format in secrets file {secrets_yaml_path}.")
if not secrets['password']:
raise ConfigError(f"Password cannot be empty in secrets file {secrets_yaml_path}.")
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']
class FileManager:
@staticmethod
def find_file(name_containing: str, with_extension: str, at_path: Path) -> Path:
return next((file for file in at_path.iterdir() if name_containing.lower() in file.name.lower() and file.suffix.lower() == with_extension.lower()), None)
@staticmethod
def validate_data_folder(app_data_folder: Path) -> tuple:
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']
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.mkdir(exist_ok=True)
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}
if resume_file:
if not resume_file.exists():
raise FileNotFoundError(f"Resume file not found: {resume_file}")
result['resume'] = resume_file
return result
def init_browser() -> webdriver.Chrome:
try:
options = chromeBrowserOptions()
service = ChromeService(ChromeDriverManager().install())
return webdriver.Chrome(service=service, options=options)
except Exception as e:
raise RuntimeError(f"Failed to initialize browser: {str(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:
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')
resume_generator_manager.choose_style()
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)
gpt_answerer_component = GPTAnswerer(openai_api_key)
bot = LinkedInBotFacade(login_component, apply_component)
bot.set_secrets(email, password)
bot.set_job_application_profile_and_resume(job_application_profile_object, resume_object)
bot.set_gpt_answerer_and_resume_generator(gpt_answerer_component, resume_generator_manager)
bot.set_parameters(parameters)
bot.start_login()
bot.start_apply()
except WebDriverException as e:
print(f"WebDriver error occurred: {e}")
except Exception as e:
raise RuntimeError(f"Error running the bot: {str(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")
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
create_and_run_bot(email, password, parameters, openai_api_key)
except ConfigError as ce:
print(f"Configuration error: {str(ce)}")
print("Refer to the configuration guide for troubleshooting: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
except FileNotFoundError as fnf:
print(f"File not found: {str(fnf)}")
print("Ensure all required files are present in the data folder.")
print("Refer to the file setup guide: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
except RuntimeError as re:
print(f"Runtime error: {str(re)}")
print("Refer to the configuration and troubleshooting guide: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
except Exception as e:
print(f"An unexpected error occurred: {str(e)}")
print("Refer to the general troubleshooting guide: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
if __name__ == "__main__":
main()