# Standard library imports import os import time from datetime import datetime from zoneinfo import ZoneInfo # Third-party imports import praw import pyfiglet import requests from dotenv import load_dotenv from flask import Flask, jsonify from threading import Thread # Load environment variables load_dotenv() # Constants REDDIT_CLIENT_ID = os.getenv('REDDIT_CLIENT_ID') REDDIT_CLIENT_SECRET = os.getenv('REDDIT_CLIENT_SECRET') REDDIT_USER_AGENT = os.getenv('REDDIT_USER_AGENT') IFTTT_WEBHOOK_URL = os.getenv('IFTTT_WEBHOOK_URL') IFTTT_WEBHOOK_URL_NIGHT = os.getenv('IFTTT_WEBHOOK_URL_NIGHT') INTERESTING_SCORE_THRESHOLD = 5 VIRAL_MULTIPLIER = 2 SCORE_INCREMENT = 5 SUBMISSION_LIMIT = 5 CHECK_INTERVAL = 300 # 5 minutes in seconds NIGHT_START_HOUR = 22 NIGHT_END_HOUR = 7 TIMEZONE = "Pacific/Auckland" MONITORED_SUBREDDITS = [ ("NintendoSwitch2", "hour"), ("nintendoswitch", "day"), ] # List to track posts we've already notified about notified_posts = [] # Initialize Reddit API client reddit = praw.Reddit( client_id=REDDIT_CLIENT_ID, client_secret=REDDIT_CLIENT_SECRET, user_agent=REDDIT_USER_AGENT, ) # Keywords that indicate potentially interesting posts interesting_words = [ "youtube", "Furukawa", "President", "Trailer", "Announcement", "Nintendo Direct", "Pre-order Trailer" ] # Official Nintendo X (Twitter) accounts to monitor nintendo_x_accounts = [ "x.com/Nintendo", "x.com/NintendoAmerica", "x.com/NintendoUK", "x.com/NintendoEU", "x.com/NintendoFrance", "x.com/NintendoGermany", "x.com/NintendoJapan", "x.com/NintendoCoLtd", ] app = Flask(__name__) @app.route('/health') def health_check(): """Health check endpoint that returns status and notified posts""" return jsonify({ 'status': 'healthy', 'notified_posts': notified_posts, 'monitored_subreddits': MONITORED_SUBREDDITS }) def run_flask(): """Run Flask server on port 5000""" app.run(host='0.0.0.0', port=5000, debug=False, threaded=True) class CommonDataObject: """ A class to process and analyze Reddit submissions for interesting content. Handles checking for keywords, X accounts, and viral metrics. """ def __init__(self, submission): """Initialize with a Reddit submission and extract key data""" self.submission = submission self.body = submission.selftext.lower() self.title = submission.title.lower() self.updoots = submission.ups self.url = submission.url.lower() def contains_nintendo_x_account(self): """Check if submission contains reference to official Nintendo X accounts""" for x_acct in nintendo_x_accounts: if x_acct.lower() in self.body or x_acct.lower() in self.url: return True return False def contains_interesting_word(self): """Check if submission contains any monitored keywords""" for word in interesting_words: if word.lower() in self.body or word.lower() in self.title: return True return False def is_viral(self, epoch_avg): """Determine if post has gone viral based on upvote comparison""" if epoch_avg > 0 and self.updoots > epoch_avg * VIRAL_MULTIPLIER: return True return False def likelihood_of_being_announcement(self, epoch_avg): """ Calculate a score indicating likelihood this is an important announcement Returns tuple of (score, reason string) """ score = 0 reason = "" if self.contains_nintendo_x_account(): score += SCORE_INCREMENT reason += "Contains Nintendo X account|" if self.contains_interesting_word(): score += SCORE_INCREMENT reason += "Contains interesting word|" if self.is_viral(epoch_avg): score += SCORE_INCREMENT reason += "Is viral|" if "youtube" in self.url or "youtube" in self.body: score += SCORE_INCREMENT reason += "Contains youtube|" return score, reason def print_splash(subreddit): """Print ASCII art title and monitoring info""" ascii_banner = pyfiglet.figlet_format("SwitchBot") print(ascii_banner) print(f"Monitoring subreddit: {subreddit}") def submission_is_interesting(submission, epoch_avg=0): """ Analyze a submission to determine if it's interesting Returns tuple of (is_interesting, score, reason) """ cdo = CommonDataObject(submission) score, reason = cdo.likelihood_of_being_announcement(epoch_avg) if score >= INTERESTING_SCORE_THRESHOLD: return True, score, reason return False, score, reason def check_reddit(subreddit_name, time_filter): """ Check a subreddit for interesting posts and process them Handles notification and IFTTT webhook calls for interesting content """ global notified_posts subreddit = reddit.subreddit(subreddit_name) print(".", end="", flush=True) submissions = list(subreddit.top(limit=SUBMISSION_LIMIT, time_filter=time_filter)) if len(submissions) == 0: return # Calculate average upvotes for this time period epoch_avg = sum(submission.ups for submission in submissions) / len(submissions) # Check each submission for interesting content for submission in submissions: interesting, score, reason = submission_is_interesting(submission, epoch_avg) if interesting: if submission.id in notified_posts: continue notified_posts.append(submission.id) print("\n" + "=" * 50) print("🎯 INTERESTING POST DETECTED! 🎯") print("=" * 50) print(f"📊 Score: {score}") print(f"🔍 Reason: {reason}") print("-" * 40) print(f"📝 Title: {submission.title}") print(f"🔗 URL: {submission.url}") print(f"📄 Content:\n{submission.selftext}") print(f"⬆️ Upvotes: {submission.ups}") print("=" * 50 + "\n") # Send to IFTTT webhook send_to_ifttt(submission, score, reason) def send_to_ifttt(submission, score, reason): """Send interesting post data to IFTTT webhook for notifications""" # Check if it's night time in NZDT (between 10pm and 7am) current_time = datetime.now(ZoneInfo(TIMEZONE)) is_night = current_time.hour >= NIGHT_START_HOUR or current_time.hour < NIGHT_END_HOUR webhook_url = IFTTT_WEBHOOK_URL_NIGHT if is_night else IFTTT_WEBHOOK_URL payload = { "value1": f"Score: {score} - {submission.title}", "value2": f"Reason: {reason}\nURL: {submission.url}", "value3": submission.selftext[:1000] if submission.selftext else "No content" } try: requests.post(webhook_url, json=payload) except Exception as e: print(f"Failed to send to IFTTT: {e}") def boot(): """Main entry point - starts monitoring loop and health check server""" print_splash(str(MONITORED_SUBREDDITS)) while True: for subreddit in MONITORED_SUBREDDITS: check_reddit(subreddit[0], subreddit[1]) time.sleep(CHECK_INTERVAL) if __name__ == "__main__": boot()