diff --git a/requirements.txt b/requirements.txt index aef4bae..341ed8e 100644 Binary files a/requirements.txt and b/requirements.txt differ diff --git a/src/linkedin-api.py b/src/linkedin-api.py new file mode 100644 index 0000000..f1395f8 --- /dev/null +++ b/src/linkedin-api.py @@ -0,0 +1,168 @@ +from typing import Dict, List +from linkedin_api import Linkedin +from typing import Optional, Union, Literal +from urllib.parse import urlencode + +class LinkedInEvolvedAPI(Linkedin): + def __init__(self, username, password): + super().__init__(username, password) + + def search_jobs( + self, + keywords: Optional[str] = None, + companies: Optional[List[str]] = None, + experience: Optional[ + List[ + Union[ + Literal["1"], + Literal["2"], + Literal["3"], + Literal["4"], + Literal["5"], + Literal["6"], + ] + ] + ] = None, + job_type: Optional[ + List[ + Union[ + Literal["F"], + Literal["C"], + Literal["P"], + Literal["T"], + Literal["I"], + Literal["V"], + Literal["O"], + ] + ] + ] = None, + job_title: Optional[List[str]] = None, + industries: Optional[List[str]] = None, + location_name: Optional[str] = None, + remote: Optional[List[Union[Literal["1"], Literal["2"], Literal["3"]]]] = None, + listed_at=24 * 60 * 60, + distance: Optional[int] = None, + easy_apply: Optional[bool] = True, + limit=-1, + offset=0, + **kwargs, + ) -> List[Dict]: + """Perform a LinkedIn search for jobs. + + :param keywords: Search keywords (str) + :type keywords: str, optional + :param companies: A list of company URN IDs (str) + :type companies: list, optional + :param experience: A list of experience levels, one or many of "1", "2", "3", "4", "5" and "6" (internship, entry level, associate, mid-senior level, director and executive, respectively) + :type experience: list, optional + :param job_type: A list of job types , one or many of "F", "C", "P", "T", "I", "V", "O" (full-time, contract, part-time, temporary, internship, volunteer and "other", respectively) + :type job_type: list, optional + :param job_title: A list of title URN IDs (str) + :type job_title: list, optional + :param industries: A list of industry URN IDs (str) + :type industries: list, optional + :param location_name: Name of the location to search within. Example: "Kyiv City, Ukraine" + :type location_name: str, optional + :param remote: Filter for remote jobs, onsite or hybrid. onsite:"1", remote:"2", hybrid:"3" + :type remote: list, optional + :param listed_at: maximum number of seconds passed since job posting. 86400 will filter job postings posted in last 24 hours. + :type listed_at: int/str, optional. Default value is equal to 24 hours. + :param distance: maximum distance from location in miles + :type distance: int/str, optional. If not specified, None or 0, the default value of 25 miles applied. + :param easy_apply: filter for jobs that are easy to apply to + :type easy_apply: bool, optional. Default value is True. + :param limit: maximum number of results obtained from API queries. -1 means maximum which is defined by constants and is equal to 1000 now. + :type limit: int, optional, default -1 + :param offset: indicates how many search results shall be skipped + :type offset: int, optional + :return: List of jobs + :rtype: list + """ + count = Linkedin._MAX_SEARCH_COUNT + if limit is None: + limit = -1 + + query: Dict[str, Union[str, Dict[str, str]]] = { + "origin": "JOB_SEARCH_PAGE_QUERY_EXPANSION" + } + if keywords: + query["keywords"] = "KEYWORD_PLACEHOLDER" + if location_name: + query["locationFallback"] = "LOCATION_PLACEHOLDER" + + query["selectedFilters"] = {} + if companies: + query["selectedFilters"]["company"] = f"List({','.join(companies)})" + if experience: + query["selectedFilters"]["experience"] = f"List({','.join(experience)})" + if job_type: + query["selectedFilters"]["jobType"] = f"List({','.join(job_type)})" + if job_title: + query["selectedFilters"]["title"] = f"List({','.join(job_title)})" + if industries: + query["selectedFilters"]["industry"] = f"List({','.join(industries)})" + if distance: + query["selectedFilters"]["distance"] = f"List({distance})" + if remote: + query["selectedFilters"]["workplaceType"] = f"List({','.join(remote)})" + if easy_apply: + query["selectedFilters"]["easyApply"] = "List(true)" + + query["selectedFilters"]["timePostedRange"] = f"List(r{listed_at})" + query["spellCorrectionEnabled"] = "true" + + query_string = ( + str(query) + .replace(" ", "") + .replace("'", "") + .replace("KEYWORD_PLACEHOLDER", keywords or "") + .replace("LOCATION_PLACEHOLDER", location_name or "") + .replace("{", "(") + .replace("}", ")") + ) + results = [] + while True: + if limit > -1 and limit - len(results) < count: + count = limit - len(results) + default_params = { + "decorationId": "com.linkedin.voyager.dash.deco.jobs.search.JobSearchCardsCollection-174", + "count": count, + "q": "jobSearch", + "query": query_string, + "start": len(results) + offset, + } + + res = self._fetch( + f"/voyagerJobsDashJobCards?{urlencode(default_params, safe='(),:')}", + headers={"accept": "application/vnd.linkedin.normalized+json+2.1"}, + ) + data = res.json() + + elements = data.get("included", []) + new_data = [] + for e in elements: + trackingUrn = e.get("trackingUrn") + if trackingUrn: + trackingUrn = trackingUrn.split(":")[-1] + e["job_id"] = trackingUrn + if e.get("$type") == "com.linkedin.voyager.dash.jobs.JobPosting": + new_data.append(e) + + if not new_data: + break + results.extend(new_data) + if ( + (-1 < limit <= len(results)) + or len(results) / count >= Linkedin._MAX_REPEATED_REQUESTS + ) or len(elements) == 0: + break + + self.logger.debug(f"results grew to {len(results)}") + + return results + + + + + + \ No newline at end of file