diff --git a/camel/toolkits/linkedin_toolkit.py b/camel/toolkits/linkedin_toolkit.py index 3ed063b53..8fc1a1ba4 100644 --- a/camel/toolkits/linkedin_toolkit.py +++ b/camel/toolkits/linkedin_toolkit.py @@ -14,14 +14,15 @@ import json import os +import warnings from http import HTTPStatus -from typing import List +from typing import List, Optional import requests from camel.toolkits import FunctionTool from camel.toolkits.base import BaseToolkit -from camel.utils import handle_http_error +from camel.utils import api_keys_required, handle_http_error LINKEDIN_POST_LIMIT = 1300 @@ -31,24 +32,29 @@ class LinkedInToolkit(BaseToolkit): This class provides methods for creating a post, deleting a post, and retrieving the authenticated user's profile information. + + Args: + linkedin_token (Optional[str]): API key for authenticating with + the Linkedin API. """ - def __init__(self): - self._access_token = self._get_access_token() + def __init__(self, linkedin_token: Optional[str] = None): + self._access_token = self._get_access_token(linkedin_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 +112,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 +142,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 +156,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,6 +169,11 @@ 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', @@ -163,10 +181,7 @@ def get_profile(self, include_id: bool = False) -> dict: "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 +209,55 @@ 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) + + if response.status_code != HTTPStatus.OK: + raise Exception( + f"Failed to get articles. " + f"Status code: {response.status_code}, " + f"Response: {response.text}" + ) + + 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,22 +270,27 @@ 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: + @api_keys_required("LINKEDIN_ACCESS_TOKEN") + def _get_access_token( + self, linkedin_token: Optional[str] = None + ) -> Optional[str]: r"""Fetches the access token required for making LinkedIn API requests. + Args: + linkedin_token (Optional[str]): API key for authenticating with + the Linkedin API. + Returns: - str: The OAuth 2.0 access token or warming message if the - environment variable `LINKEDIN_ACCESS_TOKEN` is not set or is - empty. + Optional[str]: The OAuth 2.0 access token if provided, otherwise, + returns `None`. Reference: You can apply for your personal LinkedIn API access token through the link below: https://www.linkedin.com/developers/apps """ - token = os.getenv("LINKEDIN_ACCESS_TOKEN") - if not token: - return "Access token not found. Please set LINKEDIN_ACCESS_TOKEN." + token = linkedin_token or os.getenv("LINKEDIN_ACCESS_TOKEN") return token diff --git a/test/toolkits/test_linkedin_function.py b/test/toolkits/test_linkedin_function.py index db97a0160..8b32a936a 100644 --- a/test/toolkits/test_linkedin_function.py +++ b/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