00d5babcd7
# PR Description
Adds an optional `next_token` input parameter to the
`X.SearchRecentTweetsByUsername` and `X.SearchRecentTweetsByKeywords`
tools.
This allows users to paginate through tweets. A `next_token` is provided
in the tools's response.
For example, to access the `next_token` when using the `tools.execute`,
you can do `next_token = response.output.value["meta"].get("next_token",
None)` and then pass it to the tool on your next call through the tools'
`next_token` input parameter.
125 lines
4.3 KiB
Python
125 lines
4.3 KiB
Python
from typing import Any
|
|
|
|
from arcade.sdk import ToolContext
|
|
from arcade.sdk.errors import ToolExecutionError
|
|
|
|
|
|
def get_tweet_url(tweet_id: str) -> str:
|
|
"""Get the URL of a tweet given its ID."""
|
|
return f"https://x.com/x/status/{tweet_id}"
|
|
|
|
|
|
def get_headers_with_token(context: ToolContext) -> dict[str, str]:
|
|
"""Get the headers for a request to the X API."""
|
|
if context.authorization is None or context.authorization.token is None:
|
|
raise ToolExecutionError( # noqa: TRY003
|
|
"Missing Token. Authorization is required to post a tweet.",
|
|
developer_message="Token is not set in the ToolContext.",
|
|
)
|
|
return {
|
|
"Authorization": f"Bearer {context.authorization.token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
def parse_search_recent_tweets_response(response_data: dict[str, Any]) -> dict[str, Any]:
|
|
"""
|
|
Parses response from the X API search recent tweets endpoint.
|
|
Returns the modified response data with added 'tweet_url', 'author_username', and 'author_name'.
|
|
"""
|
|
if not sanity_check_tweets_data(response_data):
|
|
return {"data": [], "next_token": ""}
|
|
|
|
# Add 'tweet_url' to each tweet
|
|
for tweet in response_data["data"]:
|
|
tweet["tweet_url"] = get_tweet_url(tweet["id"])
|
|
|
|
# Add 'author_username' and 'author_name' to each tweet
|
|
for tweet_data, user_data in zip(response_data["data"], response_data["includes"]["users"]):
|
|
tweet_data["author_username"] = user_data["username"]
|
|
tweet_data["author_name"] = user_data["name"]
|
|
|
|
return response_data
|
|
|
|
|
|
def sanity_check_tweets_data(tweets_data: dict[str, Any]) -> bool:
|
|
"""
|
|
Sanity check the tweets data.
|
|
Returns True if the tweets data is valid and contains tweets, False otherwise.
|
|
"""
|
|
if not tweets_data.get("data"):
|
|
return False
|
|
# prefer clarity over appeasing linter here
|
|
if not tweets_data.get("includes", {}).get("users"): # noqa: SIM103
|
|
return False
|
|
return True
|
|
|
|
|
|
def expand_urls_in_tweets(
|
|
tweets_data: list[dict[str, Any]], delete_entities: bool = True
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Returns a new list of tweets with expanded URLs.
|
|
"""
|
|
new_tweets = []
|
|
for tweet_data in tweets_data:
|
|
new_tweet = tweet_data.copy()
|
|
if "entities" in new_tweet and "urls" in new_tweet["entities"]:
|
|
for url_entity in new_tweet["entities"]["urls"]:
|
|
short_url = url_entity["url"]
|
|
expanded_url = url_entity["expanded_url"]
|
|
new_tweet["text"] = new_tweet["text"].replace(short_url, expanded_url)
|
|
|
|
if delete_entities:
|
|
new_tweet.pop("entities", None)
|
|
new_tweets.append(new_tweet)
|
|
return new_tweets
|
|
|
|
|
|
def expand_urls_in_user_description(user_data: dict, delete_entities: bool = True) -> dict:
|
|
"""
|
|
Returns a new user data dict with expanded URLs in the description.
|
|
"""
|
|
new_user_data = user_data.copy()
|
|
description_urls = new_user_data.get("entities", {}).get("description", {}).get("urls", [])
|
|
description = new_user_data.get("description", "")
|
|
for url_info in description_urls:
|
|
t_co_link = url_info["url"]
|
|
expanded_url = url_info["expanded_url"]
|
|
description = description.replace(t_co_link, expanded_url)
|
|
new_user_data["description"] = description
|
|
|
|
if delete_entities:
|
|
new_user_data.pop("entities", None)
|
|
return new_user_data
|
|
|
|
|
|
def expand_urls_in_user_url(user_data: dict, delete_entities: bool = True) -> dict:
|
|
"""
|
|
Returns a new user data dict with expanded URLs in the URL field.
|
|
"""
|
|
new_user_data = user_data.copy()
|
|
url_urls = new_user_data.get("entities", {}).get("url", {}).get("urls", [])
|
|
url = new_user_data.get("url", "")
|
|
for url_info in url_urls:
|
|
t_co_link = url_info["url"]
|
|
expanded_url = url_info["expanded_url"]
|
|
url = url.replace(t_co_link, expanded_url)
|
|
new_user_data["url"] = url
|
|
|
|
if delete_entities:
|
|
new_user_data.pop("entities", None)
|
|
return new_user_data
|
|
|
|
|
|
def remove_none_values(params: dict) -> dict:
|
|
"""
|
|
Remove key/value pairs with None values from a dictionary.
|
|
|
|
Args:
|
|
params: The dictionary to clean
|
|
|
|
Returns:
|
|
A new dictionary with None values removed
|
|
"""
|
|
return {k: v for k, v in params.items() if v is not None}
|