feat: allow upload of media to X

Author: celesteanglm

plugins/twitter/examples/sample_media/media_file.png

@@ -1,5 +1,6 @@
 import os
 from twitter_plugin_gamesdk.game_twitter_plugin import GameTwitterPlugin
+import base64
 
 # Define your options with the necessary credentials
 options = {
@@ -26,20 +27,44 @@
 reply_tweet_fn(tweet_id=1879472470362816626, reply="Hey! This is a test reply!", media_ids=[])
 print("Replied to tweet!")
 
-# # Test case 3: Quote a Tweet
-# print("\nRunning Test Case 3: Quote a Tweet")
-# quote_tweet_fn = game_twitter_plugin.get_function('quote_tweet')
-# quote_tweet_fn(tweet_id=1879472470362816626, quote="Hey! This is a test quote tweet!", media_ids=[])
-# print("Quoted tweet!")
-
-# # Test case 4: Search Tweets
-# print("\nRunning Test Case 4: Search Tweets")
-# search_tweets_fn = game_twitter_plugin.get_function('search_tweets')
-# response = search_tweets_fn(query="Python")
-# print(f"Searched tweets: {response}")
-
-# # Test case 5: Get authenticated user
-# print("\nRunning Test Case 5: Get details of authenticated user")
-# get_authenticated_user_fn = game_twitter_plugin.get_function('get_authenticated_user')
-# response = get_authenticated_user_fn()
-# print(f"Got details of authenticated user: {response}")
+# Test case 3: Quote a Tweet
+print("\nRunning Test Case 3: Quote a Tweet")
+quote_tweet_fn = game_twitter_plugin.get_function('quote_tweet')
+quote_tweet_fn(tweet_id=1879472470362816626, quote="Hey! This is a test quote tweet!", media_ids=[])
+print("Quoted tweet!")
+
+# Test case 4: Search Tweets
+print("\nRunning Test Case 4: Search Tweets")
+search_tweets_fn = game_twitter_plugin.get_function('search_tweets')
+response = search_tweets_fn(query="Python")
+print(f"Searched tweets: {response}")
+
+# Test case 5: Get authenticated user
+print("\nRunning Test Case 5: Get details of authenticated user")
+get_authenticated_user_fn = game_twitter_plugin.get_function('get_authenticated_user')
+response = get_authenticated_user_fn()
+print(f"Got details of authenticated user: {response}")
+
+# Test case 6: Get my mentions
+print("\nRunning Test Case 6: Get my mentions")
+mentions_fn = game_twitter_plugin.get_function('mentions')
+response = mentions_fn()
+print(f"My mentions: {response}")
+
+# Test case 7: Get my followers
+print("\nRunning Test Case 7: Get list of users who are followers of me")
+followers_fn = game_twitter_plugin.get_function('followers')
+response = followers_fn()
+print(f"My followers: {response}")
+
+# Test case 8: Get ppl following me
+print("\nRunning Test Case 8: Get list of users I am following")
+following_fn = game_twitter_plugin.get_function('following')
+response = following_fn()
+print(f"Users I am following: {response}")
+
+# Test case 9: Upload media
+print("\nRunning Test Case 9: Upload media")
+with open("sample_media/media_file.png", "rb") as f:
+    media_id = game_twitter_plugin.upload_media(f)
+    print(f"Uploaded media_id: {media_id}")

plugins/twitter/examples/test_game_twitter.py

@@ -24,7 +24,7 @@
 
 import requests
 from typing import Optional, Dict, Any, Callable, List
-import os
+import os,io
 import logging
 
 
@@ -60,6 +60,9 @@ def __init__(self, options: Dict[str, Any]) -> None:
             "quote_tweet": self._quote_tweet,
             "search_tweets": self._search_tweets,
             "get_authenticated_user": self._get_authenticated_user,
+            "mentions": self._mentions,
+            "followers": self._followers,
+            "following": self._following
         }
         # Configure logging
         logging.basicConfig(level=logging.INFO)
@@ -161,4 +164,42 @@ def _get_authenticated_user(self) -> Dict[str, Any]:
         """
         Get details of the authenticated user.
         """
-        return self._fetch_api("/me", "GET")
+        return self._fetch_api("/me", "GET")
+
+    def _mentions(self, pagination_token: Optional[str] = None) -> Dict[str, Any]:
+        """
+        Get mentions of the authenticated user.
+        """
+        endpoint = "/mentions"
+        if pagination_token:
+            endpoint += f"?paginationToken={paginationToken}"
+        return self._fetch_api(endpoint, "GET")
+
+    def _followers(self, pagination_token: Optional[str] = None) -> Dict[str, Any]:
+        """
+        Get followers of the authenticated user.
+        """
+        endpoint = "/followers"
+        if pagination_token:
+            endpoint += f"?paginationToken={paginationToken}"
+        return self._fetch_api(endpoint, "GET")
+
+    def _following(self, pagination_token: Optional[str] = None) -> Dict[str, Any]:
+        """
+        Get list of users whom the authenticated user is following.
+        """
+        endpoint = "/following"
+        if pagination_token:
+            endpoint += f"?paginationToken={paginationToken}"
+        return self._fetch_api(endpoint, "GET")
+    
+    def upload_media(self, media: bytes) -> str:
+        """
+        Uploads media (e.g. image, video) to X and returns the media ID.
+        """
+        response = requests.post(
+            url = f"{self.base_url}/media",
+            headers = {k: v for k, v in self.headers.items() if k != "Content-Type"},
+            files = {"file": media}
+        )
+        return response.json().get("mediaId")