diff --git a/requirements.txt b/requirements.txt index 5d9b670..dff7142 100644 --- a/requirements.txt +++ b/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 diff --git a/src/agent/agent.py b/src/agent/agent.py index 0201e0d..0c301d4 100644 --- a/src/agent/agent.py +++ b/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: diff --git a/src/agent/agent_tools/heygen.py b/src/agent/agent_tools/heygen.py new file mode 100644 index 0000000..9460368 --- /dev/null +++ b/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 diff --git a/src/agent/agent_tools/twitter.py b/src/agent/agent_tools/twitter.py index a9d668f..3995778 100644 --- a/src/agent/agent_tools/twitter.py +++ b/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"""