slfhstd/TestPostsBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Changes to enable always on and trigger emthod. also updated to yaml config structure

Browse Source
This commit is contained in:
Self Hosted
2026-03-11 16:36:10 +00:00
parent 827414d762
commit 1e78138992
7 changed files with 357 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot.py
+115 -12
View File
@@ -1,10 +1,11 @@
# Reddit Test Posts Bot
# This script reads config from a subreddit wiki page and makes test posts if the bot is a moderator.
# This script reads config from a subreddit wiki page and makes test posts
# when a moderator sends a chat message containing a configured trigger.
import praw
import json
import time
import os
from config import fetch_config_from_wiki
import threading
from config import fetch_config_from_wiki, validate_config_from_wiki, get_trigger_posts
REDDIT_CLIENT_ID = os.environ.get('REDDIT_CLIENT_ID')
REDDIT_CLIENT_SECRET = os.environ.get('REDDIT_CLIENT_SECRET')
@@ -17,22 +18,99 @@ WIKI_PAGE = os.environ.get('WIKI_PAGE', 'testpostsbot_config')
def is_moderator(reddit, subreddit_name):
"""Check if the bot is a moderator of the subreddit."""
subreddit = reddit.subreddit(subreddit_name)
mods = [str(mod) for mod in subreddit.moderator()]
return reddit.user.me().name in mods
def make_posts(reddit, subreddit_name, posts):
"""Make the specified posts to the subreddit."""
subreddit = reddit.subreddit(subreddit_name)
for post in posts:
title = post.get('title', 'Test Post')
body = post.get('body', '')
print(f"Posting: {title}")
subreddit.submit(title, selftext=body)
time.sleep(2) # avoid rate limits
print(f"[POSTING] Posting: {title}")
try:
subreddit.submit(title, selftext=body)
time.sleep(2) # avoid rate limits
except Exception as e:
print(f"[POSTING] Error posting '{title}': {e}")
def chat_message_watcher(reddit, subreddit_name):
"""
Watches for chat messages from moderators containing trigger keywords.
When a trigger is found, posts the configured posts for that trigger.
"""
chat_requests_file = os.path.join(os.path.dirname(__file__), 'DB', 'chat_wiki_requests.txt')
os.makedirs(os.path.dirname(chat_requests_file), exist_ok=True)
processed_message_ids = set()
# Load processed IDs from file
if os.path.exists(chat_requests_file):
with open(chat_requests_file, 'r', encoding='utf-8') as f:
for line in f:
processed_message_ids.add(line.strip())
subreddit = reddit.subreddit(subreddit_name)
while True:
try:
for message in reddit.inbox.stream():
if not hasattr(message, 'id') or message.id in processed_message_ids:
continue
processed_message_ids.add(message.id)
# Save processed ID to file
with open(chat_requests_file, 'a', encoding='utf-8') as f:
f.write(message.id + '\n')
if hasattr(message, 'body'):
message_body = message.body.lower()
# Check if sender is a moderator
author = getattr(message, 'author', None)
if author and author in subreddit.moderator():
# Load current config to check for triggers
config = fetch_config_from_wiki(reddit, subreddit_name, WIKI_PAGE)
posts_config = config.get('posts', [])
# Check if message contains any trigger
for post_config in posts_config:
if not isinstance(post_config, dict):
continue
trigger = post_config.get('trigger', '').lower()
if trigger and trigger in message_body:
print(f"[CHAT WATCH] Moderator '{author}' triggered '{trigger}'.")
# Get posts for this trigger
trigger_posts = get_trigger_posts(reddit, subreddit_name, WIKI_PAGE, trigger)
if trigger_posts:
print(f"[CHAT WATCH] Found {len(trigger_posts)} post(s) for trigger '{trigger}'.")
make_posts(reddit, subreddit_name, trigger_posts)
reply_text = f"Successfully posted {len(trigger_posts)} post(s) for trigger '{trigger}'."
else:
print(f"[CHAT WATCH] No posts found for trigger '{trigger}'.")
reply_text = f"No posts configured for trigger '{trigger}'."
try:
message.reply(reply_text)
print(f"[CHAT WATCH] Replied to chat message {message.id}.")
except Exception as e:
print(f"[CHAT WATCH] Error replying to chat message {message.id}: {e}")
# Only process the first matching trigger per message
break
except Exception as e:
print(f"[CHAT WATCH] Chat message watcher error: {e}")
time.sleep(30)
def main():
"""Main function - runs bot constantly."""
reddit = praw.Reddit(
client_id=REDDIT_CLIENT_ID,
client_secret=REDDIT_CLIENT_SECRET,
@@ -41,13 +119,38 @@ def main():
user_agent=REDDIT_USER_AGENT
)
config = fetch_config_from_wiki(reddit, SUBREDDIT, WIKI_PAGE)
posts = config.get('posts', [])
if not is_moderator(reddit, SUBREDDIT):
print(f"Bot is not a moderator of r/{SUBREDDIT}. Cannot continue.")
return
if is_moderator(reddit, SUBREDDIT):
make_posts(reddit, SUBREDDIT, posts)
else:
print(f"Bot is not a moderator of r/{SUBREDDIT}. No posts made.")
# Validate initial config
if not validate_config_from_wiki(reddit, SUBREDDIT, WIKI_PAGE):
print(f"Initial wiki config is invalid. Please fix the config at r/{SUBREDDIT}/wiki/{WIKI_PAGE}")
return
print(f"[STARTUP] TestPostsBot started for r/{SUBREDDIT}")
print(f"[STARTUP] Config page: r/{SUBREDDIT}/wiki/{WIKI_PAGE}")
print(f"[STARTUP] Bot username: {reddit.user.me()}")
# Start chat message watcher in background thread
chat_thread = threading.Thread(
target=chat_message_watcher,
args=(reddit, SUBREDDIT),
daemon=True
)
chat_thread.start()
print("[STARTUP] Chat message watcher started.")
# Keep main thread alive
try:
while True:
time.sleep(60)
except KeyboardInterrupt:
print("[SHUTDOWN] Bot shutting down...")
if __name__ == '__main__':
main()
if __name__ == '__main__':