feat: Add Linkedin get_articles

camel/toolkits/linkedin_toolkit.py

@@ -14,8 +14,9 @@
 
 import json
 import os
+import warnings
 from http import HTTPStatus
-from typing import List
+from typing import List, Optional
 
 import requests
 
@@ -36,19 +37,20 @@ class LinkedInToolkit(BaseToolkit):
     def __init__(self):
         self._access_token = self._get_access_token()
 
-    def create_post(self, text: str) -> dict:
+    def create_post(self, text: str) -> Optional[dict]:
         r"""Creates a post on LinkedIn for the authenticated user.
 
         Args:
             text (str): The content of the post to be created.
 
         Returns:
-            dict: A dictionary containing the post ID and the content of
-                the post. If the post creation fails, the values will be None.
+            Optional[dict]: A dictionary containing the post ID and the
+                content of the post. If the post creation fails, the values
+                will be :obj:`None`.
 
         Raises:
-            Exception: If the post creation fails due to
-                       an error response from LinkedIn API.
+            Exception: If the post creation fails due to an error response
+                from LinkedIn API.
         """
         url = 'https://api.linkedin.com/v2/ugcPosts'
         urn = self.get_profile(include_id=True)
@@ -106,7 +108,7 @@ def delete_post(self, post_id: str) -> str:
         Reference:
             https://docs.microsoft.com/en-us/linkedin/marketing/integrations/community-management/shares/ugc-post-api
         """
-        print(
+        warnings.warn(
             "You are going to delete a LinkedIn post "
             f"with the following ID: {post_id}"
         )
@@ -136,7 +138,11 @@ def delete_post(self, post_id: str) -> str:
 
         return f"Post deleted successfully. Post ID: {post_id}."
 
-    def get_profile(self, include_id: bool = False) -> dict:
+    def get_profile(
+        self,
+        include_id: bool = False,
+        person_id: Optional[str] = None,
+    ) -> dict:
         r"""Retrieves the authenticated user's LinkedIn profile info.
 
         This function sends a GET request to the LinkedIn API to retrieve the
@@ -146,6 +152,9 @@ def get_profile(self, include_id: bool = False) -> dict:
         Args:
             include_id (bool): Whether to include the LinkedIn profile ID in
                 the response.
+            person_id (str, optional): The ID of the member whose profile is
+                to be retrieved. If `None`, retrieves the authenticated user's
+                profile. (default: :obj:`None`)
 
         Returns:
             dict: A dictionary containing the user's LinkedIn profile
@@ -156,17 +165,19 @@ def get_profile(self, include_id: bool = False) -> dict:
             Exception: If the profile retrieval fails due to an error response
                 from LinkedIn API.
         """
+        if person_id:
+            url = f"https://api.linkedin.com/v2/people/(id:{person_id})"
+        else:
+            url = "https://api.linkedin.com/v2/userinfo"
+
         headers = {
             "Authorization": f"Bearer {self._access_token}",
             'Connection': 'Keep-Alive',
             'Content-Type': 'application/json',
             "X-Restli-Protocol-Version": "2.0.0",
         }
 
-        response = requests.get(
-            "https://api.linkedin.com/v2/userinfo",
-            headers=headers,
-        )
+        response = requests.get(url, headers=headers)
 
         if response.status_code != HTTPStatus.OK:
             raise Exception(
@@ -194,6 +205,48 @@ def get_profile(self, include_id: bool = False) -> dict:
 
         return profile_report
 
+    def get_articles(self, author_id: str) -> dict:
+        r"""Retrieves articles published by the specified author, returning
+        only essential information.
+
+        Args:
+            author_id (str): The ID of the author whose articles are to be
+                retrieved.
+
+        Returns:
+            dict: A dictionary containing the list of articles published by
+                the author.
+
+        Raises:
+            Exception: If the article retrieval fails due to an error response
+                from LinkedIn API.
+        """
+        url = f"https://api.linkedin.com/v2/originalArticles?q=authors&authors=urn:li:person:{author_id}"
+        headers = {
+            "Authorization": f"Bearer {self._access_token}",
+            'Content-Type': 'application/json',
+            "X-Restli-Protocol-Version": "2.0.0",
+        }
+
+        response = requests.get(url, headers=headers)
+
+        articles = response.json().get("elements", [])
+
+        essential_articles = []
+        for article in articles:
+            essential_info = {
+                "id": article.get("id"),
+                "title": article.get("title"),
+                "publishedAt": article.get("publishedAt"),
+                "content": article.get("content", {})
+                .get("com.linkedin.publishing.HtmlContent", {})
+                .get("htmlText"),
+                "displayImage": article.get("displayImage"),
+            }
+            essential_articles.append(essential_info)
+
+        return {"articles": essential_articles}
+
     def get_tools(self) -> List[FunctionTool]:
         r"""Returns a list of FunctionTool objects representing the
         functions in the toolkit.
@@ -206,6 +259,7 @@ def get_tools(self) -> List[FunctionTool]:
             FunctionTool(self.create_post),
             FunctionTool(self.delete_post),
             FunctionTool(self.get_profile),
+            FunctionTool(self.get_articles),
         ]
 
     def _get_access_token(self) -> str:

test/toolkits/test_linkedin_function.py

@@ -113,3 +113,44 @@ def test_get_profile(linkedin_toolkit):
             'id': 'urn:li:person:test-id',
         }
         assert response == expected_report
+
+
+def test_get_articles(linkedin_toolkit):
+    author_id = "test-id"
+    # Mock the articles response
+    mock_response = MagicMock()
+    mock_response.json.return_value = {
+        "elements": [
+            {
+                "id": "urn:li:article:test-article-id",
+                "title": "Test Article Title",
+                "publishedAt": "2024-01-01T00:00:00Z",
+                "content": {
+                    "com.linkedin.publishing.HtmlContent": {
+                        "htmlText": "This is the article content."
+                    }
+                },
+                "displayImage": "https://example.com/image.jpg",
+            }
+        ]
+    }
+    mock_response.status_code = HTTPStatus.OK
+
+    # Mock the get request
+    with patch('requests.get') as mock_get:
+        mock_get.return_value = mock_response
+
+        response = linkedin_toolkit.get_articles(author_id=author_id)
+
+        expected_response = {
+            "articles": [
+                {
+                    "id": "urn:li:article:test-article-id",
+                    "title": "Test Article Title",
+                    "publishedAt": "2024-01-01T00:00:00Z",
+                    "content": "This is the article content.",
+                    "displayImage": "https://example.com/image.jpg",
+                }
+            ]
+        }
+        assert response == expected_response