Merge pull request #349 from spectreDeveloper/v3

Added three new method to upload resume automatically.

src/linkedin-api.py

@@ -1,14 +1,13 @@
-import logging
 from typing import Dict, List
-from typing import Optional, Union, Literal
-from urllib.parse import urlencode
-
 from linkedin_api import Linkedin
+from typing import Optional, Union, Literal
+from urllib.parse import quote, urlencode, parse_qs, urlparse
+import logging
+import json
 
 # set log to all debug
 logging.basicConfig(level=logging.INFO)
 
-
 class LinkedInEvolvedAPI(Linkedin):
     already_applied_jobs: List[str] = []
     
@@ -16,44 +15,44 @@ class LinkedInEvolvedAPI(Linkedin):
         super().__init__(username, password)
 
     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]:
         """Perform a LinkedIn search for jobs.
 
@@ -160,8 +159,8 @@ class LinkedInEvolvedAPI(Linkedin):
                 break
             results.extend(new_data)
             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:
                 break
 
@@ -169,7 +168,7 @@ class LinkedInEvolvedAPI(Linkedin):
 
         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.
 
         :param job_id: Job ID
@@ -183,11 +182,13 @@ class LinkedInEvolvedAPI(Linkedin):
 
         headers: Dict[str, str] = self._headers()
         
+
         headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1"
         headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "")
         headers["Cookie"] = cookie_str
         headers["Connection"] = "keep-alive"
 
+
         default_params = {
             "decorationId": "com.linkedin.voyager.dash.deco.jobs.OnsiteApplyApplication-67",
             "jobPostingUrn": f"urn:li:fsd_jobPosting:{job_id}",
@@ -252,7 +253,7 @@ class LinkedInEvolvedAPI(Linkedin):
 
         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
     
         # ToDo: Implement apply to job parser first
@@ -268,7 +269,7 @@ class LinkedInEvolvedAPI(Linkedin):
         # EXAMPLE OF WORKING PAYLOAD
         # 4005350454 is job_id, so need to be replaced with the job_id
 
-        # {
+        #{
         #    "followCompany": true,
         #    "responses": [
         #        {
@@ -348,23 +349,161 @@ class LinkedInEvolvedAPI(Linkedin):
         #        }
         #    ],
         #    "trackingId": ""
-        # }
+        #}
 
         # Push the commit to the repository and create a pull request to the v3 branch.
         
+    def create_request_pdf(self, filename: str) -> str | None:
+        """
+        Create a PDF file with the request data.
+        :param filename: Name of the file
+        :type filename: str | None
+        :return: URL of the file uploaded to the LinkedIn.
+        """
+        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["Accept"] = "application/vnd.linkedin.normalized+json+2.1"
+        headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "")
+        headers["Cookie"] = cookie_str
+        headers["Connection"] = "keep-alive"
+
+        default_params = {
+            'action': 'requestUrl'
+        }
+
+        res = self._post(
+            f"/voyagerJobsDashAmbryUploadUrls",
+            headers=headers,
+            cookies=cookies,
+            json={"contentType":"PDF","filename":"200.pdf","maxSizeBytes":18810},
+            params=default_params
+        )
+
+
+        match res.status_code:
+            case 200:
+                parse_res = res.json()
+                url = parse_res['data']['value']
+                logging.info(url)
+                return url
+            case _:
+                self.logger.error("Failed to create a request PDF")
+                return None
+            
+    def upload_resume_via_ambry(self, url: str, cv_path: str) -> bool | str:
+        """
+        Upload resume via Ambry.
+        :param url: URL of the file uploaded to the LinkedIn.
+        :type url: str
+        :param raw_cv: Raw CV data
+        :type raw_cv: bytes
+        :return: PDF hash.pdf or false
+        """
+        binary_cv: bytes = self.file_to_binary(cv_path)
+
+        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["Accept"] = "application/vnd.linkedin.normalized+json+2.1"
+        headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "")
+        headers["Cookie"] = cookie_str
+        headers["Connection"] = "keep-alive"
+
+        ambry_url = url.replace("https://www.linkedin.com", "")
+        
+        res = self._post(
+            ambry_url,
+            base_request=True,
+            headers=headers,
+            cookies=cookies,
+            data=binary_cv,
+        )
+
+        match res.status_code:
+            case 201:
+                return res.headers['Location']
+            case _:
+                self.logger.error("Failed to upload resume via Ambry")
+                return False
+            
+    def confirm_upload_resume(self, cv_hash: str) -> bool:
+        """
+        Upload resume.
+        :param cv_hash: PDF hash
+        :type cv_hash: str
+        :return: True if success, False if failed
+        """
+
+        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["Accept"] = "application/vnd.linkedin.normalized+json+2.1"
+        headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "")
+        headers["Cookie"] = cookie_str
+        headers["Connection"] = "keep-alive"
+
+        json_data = {'entityUrn': f'urn:li:fsd_resume:{cv_hash}'}
+        res = self._post(
+            "/voyagerJobsDashResumes",
+            headers=headers,
+            cookies=cookies,
+            json=json_data,
+        )
+
+        match res.status_code:
+            case 201:
+                return True
+            case _:
+                self.logger.error("Failed to upload resume")
+                return False
+            
+    def file_to_binary(self, file_path):
+        with open(file_path, 'rb') as file:
+            binary_data = file.read()
+        return binary_data
+
     def set_job_as_applied(self, job_id: str) -> None:
         self.already_applied_jobs.append(job_id)
 
+    def upload_linkedin_resume(self, cv_path: str) -> str | bool:
+        url = self.create_request_pdf("resume.pdf")
+        if url:
+            cv_hash = self.upload_resume_via_ambry(url, cv_path)
+            if cv_hash:
+                self.confirm_upload_resume(cv_hash)
+                return cv_hash
+        return False
+            
+
+
+
 
 ## 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,
+    jobs = api.search_jobs(keywords="Frontend Developer", location_name="Italia", limit=100, easy_apply=True, offset=1, listed_at=None)
-                           listed_at=None)
     for job in jobs:
         job_id: str = job["job_id"]
-        print(f"Job ID: {job_id}")
+
-        continue
+        resume: str = api.upload_linkedin_resume("resume.pdf")
+        if isinstance(resume, bool):
+            logging.error("Failed to upload resume")
+            continue
+        elif isinstance(resume, str):
+            logging.info(f"Resume uploaded with hash {resume}")
+        else:
+            logging.error("Unknown error")
+            continue
+
 
         if job_id in api.already_applied_jobs:
             logging.info(f"Already applied to job {job_id}, skipping it")
@@ -373,4 +512,9 @@ if __name__ == "__main__":
         fields = api.get_fields_for_easy_apply(job_id)
         for field in fields:
             print(field)
+
         break
+
+        
+
+