How to Automate Reddit Trend Monitoring using the API

1#!/usr/bin/env python3
2import os
3import time
4import logging
5import requests
6from requests.auth import HTTPBasicAuth
7from collections import defaultdict, deque
8
9logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s')
10
11CLIENT_ID = os.environ['REDDIT_CLIENT_ID']
12CLIENT_SECRET = os.environ['REDDIT_CLIENT_SECRET']
13USERNAME = os.environ['REDDIT_USERNAME']
14PASSWORD = os.environ['REDDIT_PASSWORD']
15SLACK_WEBHOOK = os.environ['SLACK_WEBHOOK_URL']
16USER_AGENT = f'linux:trend-monitor:v1.0 (by /u/{USERNAME})'
17
18MONITOR_CONFIG = [
19    {'subreddit': 'MachineLearning', 'keywords': ['GPT', 'Claude', 'LLM']},
20    {'subreddit': 'technology', 'keywords': ['AI', 'automation']},
21]
22SPIKE_MULTIPLIER = 3.0
23POLL_INTERVAL = 900  # 15 minutes
24
25token = None
26token_expiry = 0
27counts_history = defaultdict(lambda: deque(maxlen=4))
28
29def get_token():
30    global token, token_expiry
31    resp = requests.post(
32        'https://www.reddit.com/api/v1/access_token',
33        auth=HTTPBasicAuth(CLIENT_ID, CLIENT_SECRET),
34        data={'grant_type': 'password', 'username': USERNAME, 'password': PASSWORD},
35        headers={'User-Agent': USER_AGENT},
36        timeout=10
37    )
38    resp.raise_for_status()
39    data = resp.json()
40    token = data['access_token']
41    token_expiry = time.time() + data['expires_in'] - 60
42    logging.info('OAuth token refreshed')
43
44def ensure_token():
45    if time.time() >= token_expiry:
46        get_token()
47
48def reddit_get(path, params=None):
49    ensure_token()
50    resp = requests.get(
51        f'https://oauth.reddit.com{path}',
52        headers={'Authorization': f'Bearer {token}', 'User-Agent': USER_AGENT},
53        params=params,
54        timeout=15
55    )
56    resp.raise_for_status()
57    data = resp.json()
58    # Check for RATELIMIT in 200 OK body
59    if isinstance(data, dict) and data.get('errors'):
60        for err in data['errors']:
61            if err[0] == 'RATELIMIT':
62                raise Exception(f'Reddit RATELIMIT: {err[1]}')
63    return data
64
65def search_subreddit(subreddit, keyword):
66    data = reddit_get(f'/r/{subreddit}/search', {
67        'q': keyword, 'restrict_sr': 1, 'sort': 'new', 'limit': 100
68    })
69    return [c['data'] for c in data['data']['children']]
70
71def count_recent(posts, minutes=15):
72    cutoff = time.time() - minutes * 60
73    return [p for p in posts if p['created_utc'] > cutoff]
74
75def send_alert(keyword, subreddit, count, baseline, top_post):
76    msg = {
77        'text': f'*Reddit Trend Alert* \u2191 "{keyword}" in r/{subreddit}: {count} mentions in 15min (baseline: {baseline:.1f}x)',
78        'attachments': [{
79            'title': top_post['title'],
80            'title_link': f'https://reddit.com{top_post["permalink"]}',
81            'text': f'Score: {top_post["score"]} | {top_post["num_comments"]} comments'
82        }]
83    }
84    requests.post(SLACK_WEBHOOK, json=msg, timeout=5)
85    logging.info(f'Alert sent: {keyword} in r/{subreddit}')
86
87def run_poll():
88    for config in MONITOR_CONFIG:
89        sub = config['subreddit']
90        for kw in config['keywords']:
91            key = f'{sub}:{kw}'
92            try:
93                posts = search_subreddit(sub, kw)
94                recent = count_recent(posts)
95                count = len(recent)
96                history = counts_history[key]
97                if len(history) >= 2:
98                    baseline = sum(history) / len(history)
99                    if baseline > 0 and count >= baseline * SPIKE_MULTIPLIER:
100                        top = max(recent, key=lambda p: p['score'], default=None)
101                        if top:
102                            send_alert(kw, sub, count, baseline, top)
103                history.append(count)
104                logging.info(f'r/{sub} "{kw}": {count} recent posts')
105                time.sleep(1)  # Respect rate limits
106            except Exception as e:
107                logging.error(f'Error checking {key}: {e}')
108
109if __name__ == '__main__':
110    get_token()
111    while True:
112        run_poll()
113        time.sleep(POLL_INTERVAL)