|
|
import logging |
|
|
import os |
|
|
import time |
|
|
from pathlib import Path |
|
|
|
|
|
import openai |
|
|
import tweepy |
|
|
|
|
|
logger = logging.getLogger() |
|
|
logging.basicConfig(level=logging.INFO) |
|
|
logger.setLevel(logging.INFO) |
|
|
|
|
|
openai_key = os.environ.get("openai_api_key") |
|
|
consumer_key = os.environ.get("consumer_key") |
|
|
consumer_secret_key = os.environ.get("consumer_secret_key") |
|
|
access_token = os.environ.get("access_token") |
|
|
access_token_secret = os.environ.get("access_token_secret") |
|
|
bearer_token = os.environ.get("bearer_token") |
|
|
|
|
|
client = tweepy.Client( |
|
|
bearer_token=bearer_token, |
|
|
consumer_key=consumer_key, |
|
|
consumer_secret=consumer_secret_key, |
|
|
access_token=access_token, |
|
|
access_token_secret=access_token_secret, |
|
|
wait_on_rate_limit=True, |
|
|
) |
|
|
|
|
|
|
|
|
def get_last_tweet(file_path: str): |
|
|
path = Path(file_path) |
|
|
if not path.exists(): |
|
|
mentions = client.get_users_mentions( |
|
|
id=1612106815096999936, |
|
|
expansions=["author_id", "in_reply_to_user_id", "referenced_tweets.id"], |
|
|
) |
|
|
since_id = mentions.data[0].id |
|
|
put_last_tweet(file_path, since_id) |
|
|
return since_id |
|
|
return int(path.read_text().strip()) |
|
|
|
|
|
|
|
|
def put_last_tweet(file_path: str, id: str): |
|
|
Path(file_path).write_text(str(id)) |
|
|
logger.info("Updated the file with the latest tweet Id") |
|
|
return |
|
|
|
|
|
|
|
|
def response(text): |
|
|
openai.api_key = openai_key |
|
|
response = openai.Completion.create( |
|
|
engine="text-davinci-002", |
|
|
prompt=text, |
|
|
temperature=0.7, |
|
|
max_tokens=60, |
|
|
top_p=1, |
|
|
frequency_penalty=0, |
|
|
presence_penalty=0, |
|
|
) |
|
|
|
|
|
return response.choices[0].text.strip() |
|
|
|
|
|
|
|
|
def roast(text): |
|
|
return response( |
|
|
f"Humorously mock, humiliate and hurt the user's feelings with a well-timed joke, diss or comeback based on the info.\n\nInfo: {text}\n\nResponse:" |
|
|
) |
|
|
|
|
|
|
|
|
def toast(text): |
|
|
return response( |
|
|
f"Give the user a genuine and unique compliment to make them feel good about themselves based on the info in a hood style manner.\n\nInfo: {text}\n\nResponse:" |
|
|
) |
|
|
|
|
|
|
|
|
def reply_to_mentions(): |
|
|
since_id = get_last_tweet("last_id.txt") |
|
|
mentions = client.get_users_mentions( |
|
|
id=1612106815096999936, |
|
|
since_id=since_id, |
|
|
expansions=["author_id", "in_reply_to_user_id", "referenced_tweets.id"], |
|
|
) |
|
|
if mentions.data is None: |
|
|
logger.info("No new mentions found") |
|
|
return |
|
|
|
|
|
for mention in reversed(mentions.data): |
|
|
try: |
|
|
if mention.author_id == 1612106815096999936: |
|
|
continue |
|
|
|
|
|
if mention.referenced_tweets is None: |
|
|
logger.info(f"Skipping {mention.id} as it is not a reply") |
|
|
tweet_to_roast_id = mention.referenced_tweets[0].id |
|
|
tweet_to_roast = client.get_tweet(id=tweet_to_roast_id) |
|
|
text_to_roast = tweet_to_roast.data.text |
|
|
|
|
|
text_out = None |
|
|
text_in = mention.text.lower() |
|
|
text_in.replace('@roastortoastgpt', '') |
|
|
text_in.replace('roastortoastgpt','') |
|
|
|
|
|
if "roast" in text_in: |
|
|
text_out = roast(text_to_roast) |
|
|
elif "toast" in text_in: |
|
|
text_out = toast(text_to_roast) |
|
|
|
|
|
if text_out is None: |
|
|
continue |
|
|
except Exception as e: |
|
|
logger.error(e) |
|
|
continue |
|
|
|
|
|
try: |
|
|
logger.info(f"Responding to: {mention.id}") |
|
|
client.create_tweet(text=text_out, in_reply_to_tweet_id=mention.id) |
|
|
except Exception as e: |
|
|
logger.error(e) |
|
|
continue |
|
|
|
|
|
put_last_tweet("last_id.txt", mention.id) |
|
|
|
|
|
|
|
|
def main(): |
|
|
while True: |
|
|
try: |
|
|
reply_to_mentions() |
|
|
except Exception as e: |
|
|
logger.error(e) |
|
|
|
|
|
import traceback |
|
|
|
|
|
traceback.print_exc() |
|
|
|
|
|
time.sleep(60) |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
main() |
|
|
|
