How to Automate Reddit Post Scheduling using the API

1import requests
2import sqlite3
3import time
4import os
5import logging
6from datetime import datetime, timedelta, timezone
7
8logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
9log = logging.getLogger('scheduler')
10
11USER_AGENT = f"python:com.example.scheduler:v1.0 (by /u/{os.environ['REDDIT_USERNAME']})"
12token_cache = {'token': None, 'expires_at': datetime.now(timezone.utc)}
13
14def get_headers():
15    if datetime.now(timezone.utc) >= token_cache['expires_at'] - timedelta(minutes=5):
16        r = requests.post(
17            'https://www.reddit.com/api/v1/access_token',
18            auth=requests.auth.HTTPBasicAuth(
19                os.environ['REDDIT_CLIENT_ID'], os.environ['REDDIT_CLIENT_SECRET']
20            ),
21            data={'grant_type': 'password',
22                  'username': os.environ['REDDIT_USERNAME'],
23                  'password': os.environ['REDDIT_PASSWORD']},
24            headers={'User-Agent': USER_AGENT}
25        )
26        r.raise_for_status()
27        data = r.json()
28        token_cache['token'] = data['access_token']
29        token_cache['expires_at'] = datetime.now(timezone.utc) + timedelta(seconds=data['expires_in'])
30    return {'Authorization': f"Bearer {token_cache['token']}", 'User-Agent': USER_AGENT}
31
32def get_due_posts(db):
33    now = datetime.now(timezone.utc).isoformat()
34    cur = db.execute(
35        "SELECT * FROM scheduled_posts WHERE scheduled_time <= ? AND status = 'pending' ORDER BY scheduled_time ASC",
36        (now,)
37    )
38    cur.row_factory = sqlite3.Row
39    return [dict(row) for row in cur.fetchall()]
40
41def submit_post(headers, post):
42    payload = {'kind': post['kind'], 'sr': post['subreddit'],
43               'title': post['title'], 'nsfw': False, 'resubmit': True}
44    if post['kind'] == 'self':
45        payload['text'] = post.get('text', '')
46    else:
47        payload['url'] = post['url']
48    if post.get('flair_id'):
49        payload['flair_id'] = post['flair_id']
50    r = requests.post('https://oauth.reddit.com/api/submit', headers=headers, data=payload)
51    r.raise_for_status()
52    result = r.json()
53    errors = result.get('json', {}).get('errors', [])
54    if errors:
55        raise Exception(f"{errors[0][0]}: {errors[0][1]}")
56    return result['json']['data']
57
58def run_scheduler(db_path='posts.db'):
59    db = sqlite3.connect(db_path)
60    log.info('Scheduler started')
61    while True:
62        try:
63            posts = get_due_posts(db)
64            for post in posts:
65                try:
66                    headers = get_headers()
67                    data = submit_post(headers, post)
68                    db.execute(
69                        "UPDATE scheduled_posts SET status='posted', reddit_post_id=? WHERE id=?",
70                        (data['id'], post['id'])
71                    )
72                    db.commit()
73                    log.info(f"Posted t3_{data['id']} to r/{post['subreddit']}: {data['url']}")
74                    time.sleep(2)
75                except Exception as e:
76                    log.error(f"Failed post {post['id']}: {e}")
77                    db.execute(
78                        "UPDATE scheduled_posts SET status='failed', error_message=? WHERE id=?",
79                        (str(e), post['id'])
80                    )
81                    db.commit()
82        except Exception as e:
83            log.error(f'Scheduler loop error: {e}')
84        time.sleep(60)
85
86if __name__ == '__main__':
87    run_scheduler()