Skip to content

post_manager

arai_ai_agents.utils.post_manager

PostManager

Manages the post manager.

Attributes:

Name Type Description
agent_name str

The name of the agent
root_folder str

The root folder
agent_folder str

The agent folder
season_folder str

The season folder
episode_folder str

The episode folder
tracker_file str

The tracker file
season_file str

The season file
episode_file str

The episode file
Source code in arai_ai_agents/utils/post_manager.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
class PostManager:
    """Manages the post manager.

    Attributes:
        agent_name (str): The name of the agent
        root_folder (str): The root folder
        agent_folder (str): The agent folder
        season_folder (str): The season folder
        episode_folder (str): The episode folder
        tracker_file (str): The tracker file
        season_file (str): The season file
        episode_file (str): The episode file
    """
    def __init__(self, agent_name: str):
        """Initialize the post manager

        Args:
            agent_name (str): The name of the agent

        Raises:
            Exception: If there's an error initializing the post manager

        Example:
            >>> post_manager = PostManager("ZorpTheAlien")
        """

        # step 1: get the twitter connector
        self.twitter_connector = twitter.TwitterConnector()

        # step 2: get the agent name
        self.agent_name = agent_name

        # step 3: get the folders
        self.root_folder = os.path.join(os.path.dirname(os.path.dirname(__file__)), "configs")
        #self.agent_folder = os.path.join(self.root_folder, "ZorpTheAlien")
        self.agent_folder = os.path.join(self.root_folder, self.agent_name)
        self.season_folder = os.path.join(self.agent_folder, "season_1")
        self.episode_folder = self.season_folder

        print (f"Agent folder: {self.agent_folder}")

        # step 4: get the tracker file
        self.tracker_file = os.path.join(self.agent_folder, "tracker.yaml")

        # Load the YAML file into a dictionary
        with open(self.tracker_file, 'r', encoding='utf-8') as f:
            self.tracker_data = yaml.safe_load(f)

        # step 5: get the season file
        self.season_file = os.path.join(self.season_folder, "season_" + str(self.tracker_data['current_season_number']) + ".yaml")

        # step 6: get the episode file
        self.episode_file = os.path.join(self.episode_folder, "s" + str(self.tracker_data['current_season_number']) + "_episode_" + str(self.tracker_data['current_episode_number']) + ".yaml")

        # step 7: load season yaml file
        with open(self.season_file, 'r', encoding='utf-8') as f:
            self.season_data = yaml.safe_load(f)

        # step 8: load episode yaml file
        # load episode yaml file
        with open(self.episode_file, 'r', encoding='utf-8') as f:
            self.episode_data = yaml.safe_load(f)

        # step 9: setup counters
        # remember that code uses indexes starting at 0
        # even if we have the number for the season, episode, and post starting at 1
        self.season_number = self.tracker_data['current_season_number']
        self.episode_number = self.tracker_data['current_episode_number']
        self.post_number = self.tracker_data['current_post_number']

        # step 10: print the counters
        print ("post_number: ", self.post_number)
        print ("episode_number: ", self.episode_number)
        print ("season_number: ", self.season_number)


    def change_season(self, season_number: int):
        """Change the season

        Args:
            season_number (int): The season number

        Example:
            >>> post_manager.change_season(1)
        """

        print(f"Changing season to ({season_number} + 1)")
        self.season_file = f"configs/{self.agent_name}/season_{season_number}/season_{season_number}.yaml"

        with open(self.season_file, 'r', encoding='utf-8') as f:
            self.season_data = yaml.safe_load(f)

        # save the tracker data
        self.tracker_data['current_season_number'] = season_number
        with open(self.tracker_file, 'w', encoding='utf-8') as f:
            yaml.dump(self.tracker_data, f)

        print(f"Changed season to {self.season_file}")

    def change_episode(self, episode_number: int):  
        """Change the episode

        Args:
            episode_number (int): The episode number

        Example:
            >>> post_manager.change_episode(1)
        """

        if episode_number >= 28-1:
            self.change_season(self.season_number + 1)            
            self.episode_number = 1

        self.episode_file = f"configs/{self.agent_name}/season_1/s1_episode_{episode_number}.yaml"

        with open(self.episode_file, 'r', encoding='utf-8') as f:
            self.episode_data = yaml.safe_load(f)

        # save the tracker data
        self.tracker_data['current_episode_number'] = episode_number
        with open(self.tracker_file, 'w', encoding='utf-8') as f:
            yaml.dump(self.tracker_data, f)

        print(f"Changed episode to {self.episode_file}")

    def next_post_number(self, post_number: int):
        """Change the post number

        Args:
            post_number (int): The post number

        Returns:
            str: The post content

        Raises:
            Exception: If there's an error changing the post number

        Example:
            >>> post_content = post_manager.change_post_number(1)
        """

        if post_number >= 12-1:
            self.change_episode(self.episode_number + 1)
            self.post_number = 0

        # Get the post content and clean it up, remember post numbers vs index starting at 0
        post_content = self.episode_data['episode']['posts'][post_number]['post_content']
        post_content = ' '.join(post_content.split()).strip()
        # Convert Unicode escape sequences to emojis
        post_content = post_content.encode().decode('unicode-escape')

        # save the tracker data
        self.tracker_data['current_post_number'] = self.post_number
        with open(self.tracker_file, 'w', encoding='utf-8') as f:
            yaml.dump(self.tracker_data, f)

        # increase post number
        self.post_number += 1        

        return post_content

    def post_to_twitter(self, live_post: bool = False):
        """Post to twitter

        Args:
            live_post (bool, optional): Whether to post to twitter. Defaults to False.

        Raises:
            Exception: If there's an error posting to twitter

        Example:
            >>> post_manager.post_to_twitter()
        """
        # print ("Preparing to post to twitter" + "\n")

        # get the next post number
        tweet_content = self.next_post_number(self.post_number)

        # if live_post is true, post to twitter
        if live_post:
            tweet_result = self.twitter_connector.post_tweet(tweet_content)

            print (tweet_result)        

            # reset the post number if the tweet failed so the post can be attempted again
            if tweet_result.startswith("Error"):
                self.post_number -= 1        

        # save to log file
        try:
            # First, read existing log if it exists
            log_file = os.path.join(self.agent_folder, f"{self.agent_name}_post_log.yaml")
            try:
                with open(log_file, 'r', encoding='utf-8') as f:
                    log_data = yaml.safe_load(f) or {'posts': []}
            except FileNotFoundError:
                log_data = {'posts': []}

            # Add new post to the array
            log_data['posts'].append({
                'post_id': f"s_{self.season_number}_e_{self.episode_number}_p_{self.post_number}",
                'content': tweet_content,
                'timestamp': str(datetime.datetime.now())
            })

            # Write back the updated log
            with open(log_file, 'w', encoding='utf-8') as f:
                yaml.dump(log_data, f, allow_unicode=True)

        except Exception as e:
            print(f"Error writing to log file: {e}")

__init__(agent_name)

Initialize the post manager

Parameters:

Name Type Description Default
agent_name str

The name of the agent

 required

Raises:

Type Description
Exception

If there's an error initializing the post manager
Example

post_manager = PostManager("ZorpTheAlien")

Source code in arai_ai_agents/utils/post_manager.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def __init__(self, agent_name: str):
    """Initialize the post manager

    Args:
        agent_name (str): The name of the agent

    Raises:
        Exception: If there's an error initializing the post manager

    Example:
        >>> post_manager = PostManager("ZorpTheAlien")
    """

    # step 1: get the twitter connector
    self.twitter_connector = twitter.TwitterConnector()

    # step 2: get the agent name
    self.agent_name = agent_name

    # step 3: get the folders
    self.root_folder = os.path.join(os.path.dirname(os.path.dirname(__file__)), "configs")
    #self.agent_folder = os.path.join(self.root_folder, "ZorpTheAlien")
    self.agent_folder = os.path.join(self.root_folder, self.agent_name)
    self.season_folder = os.path.join(self.agent_folder, "season_1")
    self.episode_folder = self.season_folder

    print (f"Agent folder: {self.agent_folder}")

    # step 4: get the tracker file
    self.tracker_file = os.path.join(self.agent_folder, "tracker.yaml")

    # Load the YAML file into a dictionary
    with open(self.tracker_file, 'r', encoding='utf-8') as f:
        self.tracker_data = yaml.safe_load(f)

    # step 5: get the season file
    self.season_file = os.path.join(self.season_folder, "season_" + str(self.tracker_data['current_season_number']) + ".yaml")

    # step 6: get the episode file
    self.episode_file = os.path.join(self.episode_folder, "s" + str(self.tracker_data['current_season_number']) + "_episode_" + str(self.tracker_data['current_episode_number']) + ".yaml")

    # step 7: load season yaml file
    with open(self.season_file, 'r', encoding='utf-8') as f:
        self.season_data = yaml.safe_load(f)

    # step 8: load episode yaml file
    # load episode yaml file
    with open(self.episode_file, 'r', encoding='utf-8') as f:
        self.episode_data = yaml.safe_load(f)

    # step 9: setup counters
    # remember that code uses indexes starting at 0
    # even if we have the number for the season, episode, and post starting at 1
    self.season_number = self.tracker_data['current_season_number']
    self.episode_number = self.tracker_data['current_episode_number']
    self.post_number = self.tracker_data['current_post_number']

    # step 10: print the counters
    print ("post_number: ", self.post_number)
    print ("episode_number: ", self.episode_number)
    print ("season_number: ", self.season_number)

change_episode(episode_number)

Change the episode

Parameters:

Name Type Description Default
episode_number int

The episode number

 required
Example

post_manager.change_episode(1)

Source code in arai_ai_agents/utils/post_manager.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def change_episode(self, episode_number: int):  
    """Change the episode

    Args:
        episode_number (int): The episode number

    Example:
        >>> post_manager.change_episode(1)
    """

    if episode_number >= 28-1:
        self.change_season(self.season_number + 1)            
        self.episode_number = 1

    self.episode_file = f"configs/{self.agent_name}/season_1/s1_episode_{episode_number}.yaml"

    with open(self.episode_file, 'r', encoding='utf-8') as f:
        self.episode_data = yaml.safe_load(f)

    # save the tracker data
    self.tracker_data['current_episode_number'] = episode_number
    with open(self.tracker_file, 'w', encoding='utf-8') as f:
        yaml.dump(self.tracker_data, f)

    print(f"Changed episode to {self.episode_file}")

change_season(season_number)

Change the season

Parameters:

Name Type Description Default
season_number int

The season number

 required
Example

post_manager.change_season(1)

Source code in arai_ai_agents/utils/post_manager.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def change_season(self, season_number: int):
    """Change the season

    Args:
        season_number (int): The season number

    Example:
        >>> post_manager.change_season(1)
    """

    print(f"Changing season to ({season_number} + 1)")
    self.season_file = f"configs/{self.agent_name}/season_{season_number}/season_{season_number}.yaml"

    with open(self.season_file, 'r', encoding='utf-8') as f:
        self.season_data = yaml.safe_load(f)

    # save the tracker data
    self.tracker_data['current_season_number'] = season_number
    with open(self.tracker_file, 'w', encoding='utf-8') as f:
        yaml.dump(self.tracker_data, f)

    print(f"Changed season to {self.season_file}")

next_post_number(post_number)

Change the post number

Parameters:

Name Type Description Default
post_number int

The post number

 required

Returns:

Name Type Description
str

The post content

Raises:

Type Description
Exception

If there's an error changing the post number
Example

post_content = post_manager.change_post_number(1)

Source code in arai_ai_agents/utils/post_manager.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def next_post_number(self, post_number: int):
    """Change the post number

    Args:
        post_number (int): The post number

    Returns:
        str: The post content

    Raises:
        Exception: If there's an error changing the post number

    Example:
        >>> post_content = post_manager.change_post_number(1)
    """

    if post_number >= 12-1:
        self.change_episode(self.episode_number + 1)
        self.post_number = 0

    # Get the post content and clean it up, remember post numbers vs index starting at 0
    post_content = self.episode_data['episode']['posts'][post_number]['post_content']
    post_content = ' '.join(post_content.split()).strip()
    # Convert Unicode escape sequences to emojis
    post_content = post_content.encode().decode('unicode-escape')

    # save the tracker data
    self.tracker_data['current_post_number'] = self.post_number
    with open(self.tracker_file, 'w', encoding='utf-8') as f:
        yaml.dump(self.tracker_data, f)

    # increase post number
    self.post_number += 1        

    return post_content

post_to_twitter(live_post=False)

Post to twitter

Parameters:

Name Type Description Default
live_post bool

Whether to post to twitter. Defaults to False.

 False

Raises:

Type Description
Exception

If there's an error posting to twitter
Example

post_manager.post_to_twitter()

Source code in arai_ai_agents/utils/post_manager.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def post_to_twitter(self, live_post: bool = False):
    """Post to twitter

    Args:
        live_post (bool, optional): Whether to post to twitter. Defaults to False.

    Raises:
        Exception: If there's an error posting to twitter

    Example:
        >>> post_manager.post_to_twitter()
    """
    # print ("Preparing to post to twitter" + "\n")

    # get the next post number
    tweet_content = self.next_post_number(self.post_number)

    # if live_post is true, post to twitter
    if live_post:
        tweet_result = self.twitter_connector.post_tweet(tweet_content)

        print (tweet_result)        

        # reset the post number if the tweet failed so the post can be attempted again
        if tweet_result.startswith("Error"):
            self.post_number -= 1        

    # save to log file
    try:
        # First, read existing log if it exists
        log_file = os.path.join(self.agent_folder, f"{self.agent_name}_post_log.yaml")
        try:
            with open(log_file, 'r', encoding='utf-8') as f:
                log_data = yaml.safe_load(f) or {'posts': []}
        except FileNotFoundError:
            log_data = {'posts': []}

        # Add new post to the array
        log_data['posts'].append({
            'post_id': f"s_{self.season_number}_e_{self.episode_number}_p_{self.post_number}",
            'content': tweet_content,
            'timestamp': str(datetime.datetime.now())
        })

        # Write back the updated log
        with open(log_file, 'w', encoding='utf-8') as f:
            yaml.dump(log_data, f, allow_unicode=True)

    except Exception as e:
        print(f"Error writing to log file: {e}")