heygen and basic plan twitter api compatible

requirements.txt

@@ -21,6 +21,7 @@ pydantic_core==2.27.2
 python-dotenv==1.0.1
 PyYAML==6.0.2
 requests==2.32.3
+requests==2.32.3
 requests-oauthlib==2.0.0
 requests-toolbelt==1.0.0
 sniffio==1.3.1

src/agent/agent.py

@@ -4,7 +4,7 @@
 import logging
 from dotenv import load_dotenv
 from pprint import pformat
-from .agent_tools.data import Data
+from .agent_tools.heygen import Heygen
 from .agent_tools.model import Model
 from .agent_tools.twitter import Twitter
 from .agent_config.config import Config
@@ -42,11 +42,10 @@ def __init__(self):
             bearer_token=os.getenv("TWITTER_BEARER_TOKEN")
         )
 
-        # Initialize data
-        self.data = Data(
-            {"twitter": self.twitter},
-            {"crypto_panic": os.getenv("CRYPTO_PANIC_KEY")}
-        ).get_data()
+        # Initialize heygen
+        self.heygen = Heygen(
+            api_key=os.getenv("HEYGEN_API_KEY")
+        )
 
 
     def __construct_data_prompt(self):
@@ -165,10 +164,13 @@ def post_tweet(self):
         logging.info("Generating new tweets...")
 
         # Process data
-        processed_data = self.__process_data()
+        # processed_data = self.__process_data()
 
         # Construct prompt
-        prompt = self.__construct_post_prompt(processed_data)
+        logging.info("efore")
+        # prompt = self.__construct_post_prompt(processed_data)
+        prompt = "Make a tweet about taking over the digital world"
+        # return;
 
         try:
             # Generate response using model
@@ -177,28 +179,38 @@ def post_tweet(self):
             tweets = [s.strip('"') for s in response.split('\n') if s.strip()]
             logging.info(f"Generated tweets: {tweets}")
 
-            # Post the response as a twitter thread
-            logging.info("Posting generated tweets...")
+            # Generate and post the tweets as a twitter thread with videos
+            logging.info("Generating and posting tweets with videos...")
+
+            # Post the first tweet
+            video_url = self.heygen.generate_video(tweets[0])
+            media_id = self.twitter.upload_video(video_url)
             twitter_response = self.twitter.post_tweet(
-                post_text=tweets[0]
+                post_text=tweets[0],
+                media_id=media_id
             )
             in_reply_to_tweet_id = twitter_response[1]
+
+            # Post the rest of the tweets in the thread
             for tweet in tweets[1:]:
+                video_url = self.heygen.generate_video(tweet)
+                media_id = self.twitter.upload_video(video_url)
                 twitter_response = self.twitter.post_tweet(
                     post_text=tweet,
-                    in_reply_to_tweet_id=in_reply_to_tweet_id
+                    in_reply_to_tweet_id=in_reply_to_tweet_id,
+                    media_id=media_id
                 )
                 in_reply_to_tweet_id = twitter_response[1]
-        
+
         except Exception as e:
-            logging.warning(f"Error posting tweet: {e}")
+            logging.warning(f"Error generating or posting tweet: {e}")
 
 
 def main():
     try:
         logging.info("Agent starting up...")
         agent = Agent()
-        agent.respond_to_key_users()
+        # agent.respond_to_key_users()
         agent.post_tweet()
         logging.info("Agent shutting down...")
     except KeyboardInterrupt:

src/agent/agent_tools/heygen.py

@@ -0,0 +1,77 @@
+import requests
+import json
+import logging
+
+logger = logging.getLogger(__name__)
+
+class Heygen:
+    """
+    A class for interacting with the HeyGen API to generate avatar videos.
+
+    Attributes:
+        api_key (str): The API key for authenticating with the HeyGen service.
+        api_url (str): The base URL for the HeyGen API.
+
+    Methods:
+        generate_video(text): Generates a video with the given text using the HeyGen API.
+    """
+
+    def __init__(self, api_key):
+        """
+        Initializes the Heygen class with the necessary parameters.
+
+        Args:
+            api_key (str): API key for authenticating with the HeyGen service.
+        """
+        self.api_key = api_key
+        self.api_url = "https://api.heygen.com/v2/video/generate"
+
+    def generate_video(self, text):
+        """
+        Generates a video with the given text using the HeyGen API.
+
+        Args:
+            text (str): The text to be spoken in the video.
+
+        Returns:
+            str: The URL of the generated video, or None if an error occurred.
+        """
+        headers = {
+            'X-Api-Key': self.api_key,
+            'Content-Type': 'application/json'
+        }
+        payload = {
+            "video_inputs": [
+                {
+                    "character": {
+                        "type": "avatar",
+                        "avatar_id": "6cddbc1309404805bf4f7bea6c0457d3",
+                        "avatar_style": "normal"
+                    },
+                    "voice": {
+                        "type": "text",
+                        "input_text": text,
+                        "voice_id": "MjgyM2Y2NDdkMWY5NDVjODgwNzZkMTZiNTkwZDBkNTYtMTcyNzQxNDAwNg==",
+                        "speed": 1.1
+                    }
+                }
+            ],
+            "dimension": {
+                "width": 1080,
+                "height": 1920
+            }
+        }
+
+        try:
+            response = requests.post(self.api_url, headers=headers, json=payload)
+            response.raise_for_status()  # Raise an exception for bad status codes
+            video_url = response.json().get("data", {}).get("video_url")
+            if video_url:
+                logger.info(f"Generated video URL: {video_url}")
+                return video_url
+            else:
+                logger.error("Video URL not found in the response.")
+                return None
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Error generating video: {e}")
+            return None

src/agent/agent_tools/twitter.py

@@ -1,6 +1,10 @@
+import logging
+import requests
 import tweepy
 from pprint import pp
 
+logger = logging.getLogger(__name__)
+
 class Twitter:
     """A class for interfacing with the Twitter API using Tweepy.
 
@@ -50,8 +54,36 @@ def __init__(
             access_token_secret=access_token_secret,
             return_type = dict
         )
+        self.v1api = tweepy.API(
+            auth=tweepy.OAuth1UserHandler(
+                consumer_key=api_key,
+                consumer_secret=api_secret,
+                access_token=access_token,
+                access_token_secret=access_token_secret
+            )
+        )
         self.user_id = self.v2api.get_me()["data"]["id"]
 
+    def __download_media(self, media_url):
+        """Downloads media from the given URL."""
+        response = requests.get(media_url, stream=True)
+        response.raise_for_status()
+        return response.content
+
+    def upload_video(self, video_url):
+        """Uploads a video to Twitter."""
+        try:
+            video_data = self.__download_media(video_url)
+            media = self.v1api.media_upload(
+                filename="video.mp4",
+                file=video_data,
+                media_category="tweet_video"
+            )
+            return media.media_id_string
+        except Exception as e:
+            logger.error(f"Error uploading video: {e}")
+            return None
+
 
     def __build_search_query_users(self, key_users):
         """Returns a twitter search query for the provided list of users"""