242 lines
9.3 KiB
Python
242 lines
9.3 KiB
Python
|
#
|
||
|
|
# Module: post_manager
|
||
|
|
#
|
||
|
|
# This module implements the PostManager class for managing the posts for an agent.
|
||
|
|
#
|
||
|
|
# Title: Post Manager
|
||
|
|
# Summary: Post manager implementation.
|
||
|
|
# Authors:
|
||
|
|
# - @TheBlockRhino
|
||
|
|
# Created: 2024-12-31
|
||
|
|
# Last edited by: @TheBlockRhino
|
||
|
|
# Last edited date: 2025-01-04
|
||
|
|
# URLs:
|
||
|
|
# - https://arai-ai.io
|
||
|
|
# - https://github.com/ARAI-DevHub/arai-ai-agents
|
||
|
|
# - https://x.com/TheBlockRhino
|
||
|
|
|
||
|
|
# standard imports
|
||
|
|
import os
|
||
|
|
import json
|
||
|
|
import datetime
|
||
|
|
from dotenv import load_dotenv
|
||
|
|
import sys
|
||
|
|
|
||
|
|
# Add project root to Python path
|
||
|
|
project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
|
sys.path.append(project_root)
|
||
|
|
|
||
|
|
# custom ARAI code imports
|
||
|
|
import connectors.twitter_connector as twitter
|
||
|
|
from connectors.twitter_api_free_connector import (
|
||
|
|
login_and_save_state,
|
||
|
|
post_tweet_with_saved_state
|
||
|
|
)
|
||
|
|
from handlers import menu_handlers
|
||
|
|
|
||
|
|
# Load environment variables
|
||
|
|
load_dotenv()
|
||
|
|
|
||
|
|
# Define constants from environment variables
|
||
|
|
X_ENABLED = os.getenv("X_ENABLED", "False")
|
||
|
|
X_API_OFFICIAL = os.getenv("X_API_OFFICIAL", "False")
|
||
|
|
X_USERNAME = os.getenv("X_USERNAME", "")
|
||
|
|
X_PASSWORD = os.getenv("X_PASSWORD","")
|
||
|
|
X_PHONE_OR_EMAIL = os.getenv("X_PHONE_OR_EMAIL", "")
|
||
|
|
X_DRY_RUN = os.getenv("X_DRY_RUN", "True")
|
||
|
|
|
||
|
|
class PostManager:
|
||
|
|
"""Manages posts for an agent using JSON configuration files.
|
||
|
|
|
||
|
|
Attributes:
|
||
|
|
agent_name (str): The name of the agent
|
||
|
|
master_file (str): Path to the agent's master JSON file
|
||
|
|
twitter_connector: Twitter API connector instance
|
||
|
|
"""
|
||
|
|
def __init__(self, agent_name: str):
|
||
|
|
"""Initialize the post manager
|
||
|
|
|
||
|
|
Args:
|
||
|
|
agent_name (str): The name of the agent
|
||
|
|
|
||
|
|
Raises:
|
||
|
|
FileNotFoundError: If master JSON file doesn't exist
|
||
|
|
"""
|
||
|
|
self.agent_name = agent_name
|
||
|
|
self.master_file = os.path.join("configs", agent_name, f"{agent_name}_master.json")
|
||
|
|
self.is_logged_in = False
|
||
|
|
|
||
|
|
# Initialize Twitter connectors based on environment variable
|
||
|
|
if X_API_OFFICIAL == "True":
|
||
|
|
print(f"[Post Manager] Initializing Twitter Connector with official API")
|
||
|
|
self.twitter_connector = twitter.TwitterConnector()
|
||
|
|
self.twitter_api_free = None
|
||
|
|
else:
|
||
|
|
print(f"[Post Manager] Initializing Twitter API Free")
|
||
|
|
self.twitter_connector = None
|
||
|
|
self.twitter_api_free = login_and_save_state(username=X_USERNAME, password=X_PASSWORD, phone_or_email=X_PHONE_OR_EMAIL)
|
||
|
|
if self.twitter_api_free:
|
||
|
|
self.is_logged_in = True
|
||
|
|
|
||
|
|
# Load master configuration
|
||
|
|
with open(self.master_file, 'r', encoding='utf-8') as f:
|
||
|
|
self.master_data = json.load(f)
|
||
|
|
|
||
|
|
# Get current tracking info
|
||
|
|
self.tracker = self.master_data["agent"]["tracker"]
|
||
|
|
self.current_season = self.tracker["current_season_number"]
|
||
|
|
self.current_episode = self.tracker["current_episode_number"]
|
||
|
|
self.current_post = self.tracker["current_post_number"]
|
||
|
|
|
||
|
|
def _save_master_file(self):
|
||
|
|
"""Save the master JSON file with updated tracking information"""
|
||
|
|
# Update master data from local variables
|
||
|
|
self.master_data["agent"]["tracker"]["current_season_number"] = self.current_season
|
||
|
|
self.master_data["agent"]["tracker"]["current_episode_number"] = self.current_episode
|
||
|
|
self.master_data["agent"]["tracker"]["current_post_number"] = self.current_post
|
||
|
|
|
||
|
|
# Save to file
|
||
|
|
with open(self.master_file, 'w', encoding='utf-8') as f:
|
||
|
|
json.dump(self.master_data, f, indent=2)
|
||
|
|
|
||
|
|
def change_season(self, season_number: int):
|
||
|
|
"""Change the current season
|
||
|
|
|
||
|
|
Args:
|
||
|
|
season_number (int): The season number (1-based, will be converted to 0-based for array access)
|
||
|
|
"""
|
||
|
|
# Convert from 1-based season number to 0-based array index
|
||
|
|
season_index = season_number
|
||
|
|
|
||
|
|
# Check if season exists and is not posted yet
|
||
|
|
if 0 <= season_index < len(self.master_data["agent"]["seasons"]):
|
||
|
|
print(f"Checking season: {season_index}")
|
||
|
|
season = self.master_data["agent"]["seasons"][season_index]
|
||
|
|
if not season.get("season_posted", False):
|
||
|
|
self.current_season = season_index # Store as 0-based index
|
||
|
|
self.current_episode = 0
|
||
|
|
self.current_post = 0
|
||
|
|
self._save_master_file()
|
||
|
|
print(f"Changed to season {season_number}") # Display as 1-based number
|
||
|
|
else:
|
||
|
|
print(f"Season {season_number} has already been posted")
|
||
|
|
raise ValueError(f"Season {season_number} has already been posted")
|
||
|
|
else:
|
||
|
|
print("No more unposted seasons available")
|
||
|
|
# Reset to beginning
|
||
|
|
self.current_season = -1
|
||
|
|
self.current_episode = -1
|
||
|
|
self.current_post = -1
|
||
|
|
self._save_master_file()
|
||
|
|
raise ValueError(f"No more content available to post")
|
||
|
|
|
||
|
|
def change_episode(self, episode_number: int):
|
||
|
|
"""Change the current episode
|
||
|
|
|
||
|
|
Args:
|
||
|
|
episode_number (int): The episode number (1-based, will be converted to 0-based for array access)
|
||
|
|
"""
|
||
|
|
# Convert from 1-based episode number to 0-based array index
|
||
|
|
episode_index = episode_number
|
||
|
|
|
||
|
|
current_season = self.master_data["agent"]["seasons"][self.current_season]
|
||
|
|
if episode_index >= len(current_season["episodes"]):
|
||
|
|
print(f"Moving to next season: {self.current_season + 1}")
|
||
|
|
self.change_season(self.current_season + 1) # Move to next season
|
||
|
|
return
|
||
|
|
|
||
|
|
self.current_episode = episode_index
|
||
|
|
self.current_post = 0
|
||
|
|
self._save_master_file()
|
||
|
|
print(f"Changed to episode {episode_number}") # Display as 1-based number
|
||
|
|
|
||
|
|
def next_post_number(self, post_number: int) -> str:
|
||
|
|
"""Get the next post content and update tracking
|
||
|
|
|
||
|
|
Args:
|
||
|
|
post_number (int): The post number (1-based, will be converted to 0-based for array access)
|
||
|
|
"""
|
||
|
|
# Convert from 1-based post number to 0-based array index
|
||
|
|
post_index = post_number
|
||
|
|
|
||
|
|
current_season = self.master_data["agent"]["seasons"][self.current_season]
|
||
|
|
current_episode = current_season["episodes"][self.current_episode]
|
||
|
|
|
||
|
|
if post_index >= len(current_episode["posts"]):
|
||
|
|
print(f"Moving to next episode: {self.current_episode + 1}")
|
||
|
|
self.change_episode(self.current_episode + 1) # Move to next episode
|
||
|
|
post_index = 0
|
||
|
|
|
||
|
|
print(f"Getting post: {post_index}")
|
||
|
|
post = current_episode["posts"][post_index]
|
||
|
|
post_content = post["post_content"]
|
||
|
|
|
||
|
|
self.current_post = post_index + 1 # Store as 1-based for consistency with display
|
||
|
|
self._save_master_file()
|
||
|
|
|
||
|
|
return post_content
|
||
|
|
|
||
|
|
def post_to_twitter(self):
|
||
|
|
"""Post content to Twitter"""
|
||
|
|
tweet_content = self.next_post_number(self.current_post)
|
||
|
|
|
||
|
|
if X_ENABLED:
|
||
|
|
if X_API_OFFICIAL:
|
||
|
|
if self.twitter_connector and not X_DRY_RUN:
|
||
|
|
tweet_result = self.twitter_connector.post_tweet(tweet_content)
|
||
|
|
else:
|
||
|
|
print(f"[Post Manager] - Dry run mode, not posting to Twitter", tweet_content)
|
||
|
|
else:
|
||
|
|
if self.twitter_api_free and not X_DRY_RUN:
|
||
|
|
tweet_result = post_tweet_with_saved_state(tweet_content)
|
||
|
|
else:
|
||
|
|
print(f"[Post Manager] - Dry run mode, not posting to Twitter", tweet_content)
|
||
|
|
|
||
|
|
|
||
|
|
print(f"[Post Manager] Tweet result: {tweet_result}")
|
||
|
|
|
||
|
|
if tweet_result.startswith("Error"):
|
||
|
|
self.current_post -= 1
|
||
|
|
self.tracker["current_post_number"] = self.current_post
|
||
|
|
self._save_master_file()
|
||
|
|
else:
|
||
|
|
print(f"\nWould have posted: \n{tweet_content}")
|
||
|
|
|
||
|
|
# Log the post
|
||
|
|
log_file = os.path.join("configs", self.agent_name, f"{self.agent_name}_post_log.json")
|
||
|
|
try:
|
||
|
|
try:
|
||
|
|
with open(log_file, 'r', encoding='utf-8') as f:
|
||
|
|
log_data = json.load(f)
|
||
|
|
except FileNotFoundError:
|
||
|
|
log_data = {"posts": []}
|
||
|
|
|
||
|
|
log_data["posts"].append({
|
||
|
|
"post_id": f"s_{self.current_season + 1}_e_{self.current_episode + 1}_p_{self.current_post}",
|
||
|
|
"content": tweet_content,
|
||
|
|
"timestamp": datetime.datetime.now().isoformat()
|
||
|
|
})
|
||
|
|
|
||
|
|
with open(log_file, 'w', encoding='utf-8') as f:
|
||
|
|
json.dump(log_data, f, indent=2)
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error writing to log file: {e}")
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def post_single_tweet(cls, tweet_content: str) -> str:
|
||
|
|
"""Class method to post a single tweet without needing to instantiate PostManager
|
||
|
|
|
||
|
|
Args:
|
||
|
|
tweet_content (str): The content to tweet
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
str: Result of the tweet operation
|
||
|
|
"""
|
||
|
|
if X_ENABLED == "True" and X_DRY_RUN == "False":
|
||
|
|
return post_tweet_with_saved_state(tweet_content)
|
||
|
|
else:
|
||
|
|
print(f"[Post Manager] - Dry run mode, would have posted: {tweet_content}")
|
||
|
|
return "Dry run - tweet not posted"
|
||
|
|
|