added resume upload

This commit is contained in:
Manu Altieri 2024-09-10 21:25:19 +02:00
parent 40cf3e4d34
commit d9ffc7542c

View file

@ -1,66 +1,58 @@
<<<<<<< HEAD
from typing import Dict, List from typing import Dict, List
from linkedin_api import Linkedin from linkedin_api import Linkedin
from typing import Optional, Union, Literal from typing import Optional, Union, Literal
from urllib.parse import quote, urlencode, parse_qs, urlparse from urllib.parse import quote, urlencode, parse_qs, urlparse
=======
>>>>>>> upstream/v3
import logging import logging
from typing import Dict, List import json
from typing import Optional, Union, Literal
from urllib.parse import urlencode
from linkedin_api import Linkedin
# set log to all debug # set log to all debug
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
class LinkedInEvolvedAPI(Linkedin): class LinkedInEvolvedAPI(Linkedin):
already_applied_jobs: List[str] = [] already_applied_jobs: List[str] = []
def __init__(self, username, password): def __init__(self, username, password):
super().__init__(username, password) super().__init__(username, password)
def search_jobs( def search_jobs(
self, self,
keywords: Optional[str] = None, keywords: Optional[str] = None,
companies: Optional[List[str]] = None, companies: Optional[List[str]] = None,
experience: Optional[ experience: Optional[
List[ List[
Union[ Union[
Literal["1"], Literal["1"],
Literal["2"], Literal["2"],
Literal["3"], Literal["3"],
Literal["4"], Literal["4"],
Literal["5"], Literal["5"],
Literal["6"], Literal["6"],
]
] ]
] = None, ]
job_type: Optional[ ] = None,
List[ job_type: Optional[
Union[ List[
Literal["F"], Union[
Literal["C"], Literal["F"],
Literal["P"], Literal["C"],
Literal["T"], Literal["P"],
Literal["I"], Literal["T"],
Literal["V"], Literal["I"],
Literal["O"], Literal["V"],
] Literal["O"],
] ]
] = None, ]
job_title: Optional[List[str]] = None, ] = None,
industries: Optional[List[str]] = None, job_title: Optional[List[str]] = None,
location_name: Optional[str] = None, industries: Optional[List[str]] = None,
remote: Optional[List[Union[Literal["1"], Literal["2"], Literal["3"]]]] = None, location_name: Optional[str] = None,
listed_at: None | int = None, remote: Optional[List[Union[Literal["1"], Literal["2"], Literal["3"]]]] = None,
distance: Optional[int] = None, listed_at: None | int = None,
easy_apply: Optional[bool] = True, distance: Optional[int] = None,
limit=-1, easy_apply: Optional[bool] = True,
offset=0, limit=-1,
**kwargs, offset=0,
**kwargs,
) -> List[Dict]: ) -> List[Dict]:
"""Perform a LinkedIn search for jobs. """Perform a LinkedIn search for jobs.
@ -162,21 +154,21 @@ class LinkedInEvolvedAPI(Linkedin):
e["job_id"] = trackingUrn e["job_id"] = trackingUrn
if e.get("$type") == "com.linkedin.voyager.dash.jobs.JobPosting": if e.get("$type") == "com.linkedin.voyager.dash.jobs.JobPosting":
new_data.append(e) new_data.append(e)
if not new_data: if not new_data:
break break
results.extend(new_data) results.extend(new_data)
if ( if (
(-1 < limit <= len(results)) (-1 < limit <= len(results))
or len(results) / count >= Linkedin._MAX_REPEATED_REQUESTS or len(results) / count >= Linkedin._MAX_REPEATED_REQUESTS
) or len(elements) == 0: ) or len(elements) == 0:
break break
self.logger.debug(f"results grew to {len(results)}") self.logger.debug(f"results grew to {len(results)}")
return results return results
def get_fields_for_easy_apply(self, job_id: str) -> List[Dict]: def get_fields_for_easy_apply(self,job_id: str) -> List[Dict]:
"""Get fields needed for easy apply jobs. """Get fields needed for easy apply jobs.
:param job_id: Job ID :param job_id: Job ID
@ -189,12 +181,14 @@ class LinkedInEvolvedAPI(Linkedin):
cookie_str = "; ".join([f"{k}={v}" for k, v in cookies.items()]) cookie_str = "; ".join([f"{k}={v}" for k, v in cookies.items()])
headers: Dict[str, str] = self._headers() headers: Dict[str, str] = self._headers()
headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1" headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1"
headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "") headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "")
headers["Cookie"] = cookie_str headers["Cookie"] = cookie_str
headers["Connection"] = "keep-alive" headers["Connection"] = "keep-alive"
default_params = { default_params = {
"decorationId": "com.linkedin.voyager.dash.deco.jobs.OnsiteApplyApplication-67", "decorationId": "com.linkedin.voyager.dash.deco.jobs.OnsiteApplyApplication-67",
"jobPostingUrn": f"urn:li:fsd_jobPosting:{job_id}", "jobPostingUrn": f"urn:li:fsd_jobPosting:{job_id}",
@ -223,26 +217,26 @@ class LinkedInEvolvedAPI(Linkedin):
except ValueError: except ValueError:
self.logger.error("Failed to parse JSON response") self.logger.error("Failed to parse JSON response")
return [] return []
form_components = [] form_components = []
for item in data.get("included", []): for item in data.get("included", []):
if 'formComponent' in item: if 'formComponent' in item:
urn = item['urn'] urn = item['urn']
try: try:
title = item['title']['text'] title = item['title']['text']
except TypeError: except TypeError:
title = urn title = urn
form_component_type = list(item['formComponent'].keys())[0] form_component_type = list(item['formComponent'].keys())[0]
form_component_details = item['formComponent'][form_component_type] form_component_details = item['formComponent'][form_component_type]
component_info = { component_info = {
'title': title, 'title': title,
'urn': urn, 'urn': urn,
'formComponentType': form_component_type, 'formComponentType': form_component_type,
} }
if 'textSelectableOptions' in form_component_details: if 'textSelectableOptions' in form_component_details:
options = [ options = [
opt['optionText']['text'] for opt in form_component_details['textSelectableOptions'] opt['optionText']['text'] for opt in form_component_details['textSelectableOptions']
@ -250,18 +244,18 @@ class LinkedInEvolvedAPI(Linkedin):
component_info['selectableOptions'] = options component_info['selectableOptions'] = options
elif 'selectableOptions' in form_component_details: elif 'selectableOptions' in form_component_details:
options = [ options = [
opt['textSelectableOption']['optionText']['text'] opt['textSelectableOption']['optionText']['text']
for opt in form_component_details['selectableOptions'] for opt in form_component_details['selectableOptions']
] ]
component_info['selectableOptions'] = options component_info['selectableOptions'] = options
form_components.append(component_info) form_components.append(component_info)
return form_components return form_components
def apply_to_job(self, job_id: str, fields: dict, followCompany: bool = True) -> bool: def apply_to_job(self,job_id: str, fields: dict, followCompany: bool = True) -> bool:
return False return False
# ToDo: Implement apply to job parser first # ToDo: Implement apply to job parser first
# How need to be implemented: # How need to be implemented:
# 1. Get fields for easy apply job from the previous method (get_fields_for_easy_apply) # 1. Get fields for easy apply job from the previous method (get_fields_for_easy_apply)
@ -271,11 +265,11 @@ class LinkedInEvolvedAPI(Linkedin):
# {'title': 'Quanti anni di esperienza di lavoro hai con Router?', 'urn': 'urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4013860791,9478711764,numeric)', 'formComponentType': 'singleLineTextFormComponent', 'response': '5'} # {'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) # 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. # Build a working payload.
# EXAMPLE OF WORKING PAYLOAD # EXAMPLE OF WORKING PAYLOAD
# 4005350454 is job_id, so need to be replaced with the job_id # 4005350454 is job_id, so need to be replaced with the job_id
# { #{
# "followCompany": true, # "followCompany": true,
# "responses": [ # "responses": [
# { # {
@ -355,10 +349,9 @@ class LinkedInEvolvedAPI(Linkedin):
# } # }
# ], # ],
# "trackingId": "" # "trackingId": ""
# } #}
# Push the commit to the repository and create a pull request to the v3 branch. # Push the commit to the repository and create a pull request to the v3 branch.
<<<<<<< HEAD
def create_request_pdf(self, filename: str) -> str | None: def create_request_pdf(self, filename: str) -> str | None:
""" """
@ -476,13 +469,10 @@ class LinkedInEvolvedAPI(Linkedin):
with open(file_path, 'rb') as file: with open(file_path, 'rb') as file:
binary_data = file.read() binary_data = file.read()
return binary_data return binary_data
=======
>>>>>>> upstream/v3
def set_job_as_applied(self, job_id: str) -> None: def set_job_as_applied(self, job_id: str) -> None:
self.already_applied_jobs.append(job_id) self.already_applied_jobs.append(job_id)
<<<<<<< HEAD
def upload_linkedin_resume(self, cv_path: str) -> str | bool: def upload_linkedin_resume(self, cv_path: str) -> str | bool:
url = self.create_request_pdf("resume.pdf") url = self.create_request_pdf("resume.pdf")
if url: if url:
@ -501,14 +491,6 @@ if __name__ == "__main__":
api: LinkedInEvolvedAPI = LinkedInEvolvedAPI(username="", password="") 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)
=======
## 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)
>>>>>>> upstream/v3
for job in jobs: for job in jobs:
job_id: str = job["job_id"] job_id: str = job["job_id"]
@ -532,3 +514,7 @@ if __name__ == "__main__":
print(field) print(field)
break break