From c0f186ba844f7b12ea515675709c4977bc817e32 Mon Sep 17 00:00:00 2001 From: Manu Altieri Date: Sun, 1 Sep 2024 22:48:01 +0200 Subject: [PATCH 1/3] Fixed easy apply search, implemented method to extract required fields from API with new endpoint, Todo; parse result json to return a dict that contains title, optionUrn and possible choices for the value if present --- src/linkedin-api.py | 76 +++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 73 insertions(+), 3 deletions(-) diff --git a/src/linkedin-api.py b/src/linkedin-api.py index f1395f8..6bb2c12 100644 --- a/src/linkedin-api.py +++ b/src/linkedin-api.py @@ -1,12 +1,19 @@ 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): def __init__(self, username, password): super().__init__(username, password) + def search_jobs( self, keywords: Optional[str] = None, @@ -106,7 +113,7 @@ 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})" query["spellCorrectionEnabled"] = "true" @@ -160,9 +167,72 @@ 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['User-Agent'] = headers['user-agent'] + headers['Accept-Language'] = headers['accept-language'] + headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1" + headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "") + headers["Cookie"] = cookie_str + headers["Connection"] = "keep-alive" + + headers.pop("user-agent") + headers.pop("accept-language") + 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, + ) + + 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: + title = item['title']['text'] + form_components.append({title: item['formComponent']}) + + + + + def get_cookies_hitting_url(self, url: str): + res = self._fetch(url,base_request=False) + return res.headers + +## EXAMPLE USAGE +#if __name__ == "__main__": +# api: LinkedInEvolvedAPI = LinkedInEvolvedAPI("", "") +# jobs = api.search_jobs(keywords="Python", location_name="Italy", limit=1, easy_apply=True) +# for job in jobs: +# job_id: str = job["job_id"] +# fields = api.get_fields_for_easy_apply(job_id) + \ No newline at end of file From d76a35fa7ca0d53f93035ca8ed5e2f314b5b424d Mon Sep 17 00:00:00 2001 From: Manu Altieri Date: Sun, 1 Sep 2024 22:53:05 +0200 Subject: [PATCH 2/3] Fixed easy apply search, implemented method to extract required fields from API with new endpoint, Todo; parse result json to return a dict that contains title, optionUrn and possible choices for the value if present --- src/linkedin-api.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/src/linkedin-api.py b/src/linkedin-api.py index 6bb2c12..8b642ce 100644 --- a/src/linkedin-api.py +++ b/src/linkedin-api.py @@ -222,10 +222,6 @@ class LinkedInEvolvedAPI(Linkedin): - def get_cookies_hitting_url(self, url: str): - res = self._fetch(url,base_request=False) - return res.headers - ## EXAMPLE USAGE #if __name__ == "__main__": # api: LinkedInEvolvedAPI = LinkedInEvolvedAPI("", "") From d2372523fa876c3abbceffd9c7b371a3afb10535 Mon Sep 17 00:00:00 2001 From: Manu Altieri Date: Mon, 2 Sep 2024 11:10:18 +0200 Subject: [PATCH 3/3] fixed the parsing of the required fields for apply --- src/linkedin-api.py | 66 ++++++++++++++++++++++++++++++++++----------- 1 file changed, 50 insertions(+), 16 deletions(-) diff --git a/src/linkedin-api.py b/src/linkedin-api.py index 8b642ce..a34e381 100644 --- a/src/linkedin-api.py +++ b/src/linkedin-api.py @@ -13,7 +13,6 @@ class LinkedInEvolvedAPI(Linkedin): def __init__(self, username, password): super().__init__(username, password) - def search_jobs( self, keywords: Optional[str] = None, @@ -182,16 +181,12 @@ class LinkedInEvolvedAPI(Linkedin): headers: Dict[str, str] = self._headers() - headers['User-Agent'] = headers['user-agent'] - headers['Accept-Language'] = headers['accept-language'] + headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1" headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "") headers["Cookie"] = cookie_str headers["Connection"] = "keep-alive" - headers.pop("user-agent") - headers.pop("accept-language") - default_params = { "decorationId": "com.linkedin.voyager.dash.deco.jobs.OnsiteApplyApplication-67", @@ -206,6 +201,16 @@ class LinkedInEvolvedAPI(Linkedin): 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: @@ -216,19 +221,48 @@ class LinkedInEvolvedAPI(Linkedin): for item in data.get("included", []): if 'formComponent' in item: - title = item['title']['text'] - form_components.append({title: item['formComponent']}) + 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 - - ## EXAMPLE USAGE -#if __name__ == "__main__": -# api: LinkedInEvolvedAPI = LinkedInEvolvedAPI("", "") -# jobs = api.search_jobs(keywords="Python", location_name="Italy", limit=1, easy_apply=True) -# for job in jobs: -# job_id: str = job["job_id"] -# fields = api.get_fields_for_easy_apply(job_id) +if __name__ == "__main__": + api: LinkedInEvolvedAPI = LinkedInEvolvedAPI(username="", password="") + jobs = api.search_jobs(keywords="Frontend Developer", location_name="Italia", limit=5, easy_apply=True, offset=1) + for job in jobs: + job_id: str = job["job_id"] + + fields = api.get_fields_for_easy_apply(job_id) + for field in fields: + print(field) + \ No newline at end of file