Skip to content

Instantly share code, notes, and snippets.

# bot.py
def fetch_submission_ids(self, subreddit, **kwargs):
return [submission.id for submission in self.reddit_client.subreddit(subreddit).hot(**kwargs)]
def parse_submission(self, submission_id=None, url=None):
try:
submission = self.reddit_client.submission(id=submission_id, url=url)
self._process_submission(submission)
except Exception as ex:
error_message = str(ex)
Sun 26 16:32:01 MainThread INFO My-bot - Processed 'help a brother out'.
Sun 26 16:32:05 MainThread INFO My-bot - Processed 'Please I want to be able to post my dank memes'.
Sun 26 16:32:12 MainThread INFO My-bot - Processed 'Free sushi to all who upd00t! Leave 1, Get 1!'. Upvoted and replied to 1 comments
Sun 26 16:32:16 MainThread INFO My-bot - Processed 'Here’s the blank meme format'.
Sun 26 16:32:19 MainThread INFO My-bot - Processed 'I need some karma to post memes'.
def _process_comments(self, comments):
num_of_replies = round(len(comments) * 0.35)
comments_to_leave = iter(random.sample(comment_reply_list, num_of_replies + 1))
for comment in random.sample(comments, num_of_replies):
comment.upvote()
sleep(random.randint(2, 3))
comment.reply(next(comments_to_leave, random.choice(comment_reply_list)))
return num_of_replies
def _process_submission(self, submission):
...
if submission_id not in self.passed_submissions:
...
self.passed_submissions.add(submission_id)
comments_to_ignore = self._compute_comments_to_ignore(comments)
comment_count = None
if not comments_to_ignore:
comment_count = self._process_comments(comments)
def _compute_comments_to_ignore(self, comments) -> dict:
to_ignore = {}
for comment in comments:
if comment.author and comment.author.name == self.username:
to_ignore[comment.id] = comment
for reply in comment.replies:
if reply.author and reply.author.name == self.username:
to_ignore[reply.id] = reply
return to_ignore
# in bot.py
def _retry_rate_limited_failure(self, error_msg, func, *args, **kwargs):
error_msg = error_msg.lower()
search_term = "try again in "
if search_term in error_msg:
minute_idx = error_msg.index(search_term) + len(search_term)
if error_msg[minute_idx].isdigit():
digits = [error_msg[minute_idx]]
if error_msg[minute_idx + 1].isdigit():
digits.append(error_msg[minute_idx + 1])
try:
...
except Exception as ex:
error_message = str(ex)
self._retry_rate_limited_failure(error_message, self.work_on_subreddit, subreddit, **generator_kwargs)
# main.py
import json
import logging
import os
from pathlib import Path
from bot import RedditBot
logging.basicConfig(handlers=[logging.StreamHandler()],
level=logging.INFO,
# bot.py
import logging
import random
import praw
from store import dump_pickled, read_pickled_set
submission_reply_list = ["Nice", "up up", "nice job, friend",
"I like your creativity", "Very nice"]
@dev4Fun
dev4Fun / block9.py
Last active February 7, 2020 04:25
def work_on_subreddit(self, subreddit: str, **generator_kwargs):
...
try:
submissions = list(self.reddit_client.subreddit(subreddit).new(**generator_kwargs))
...
finally:
dump_pickled(self.passed_submissions, f"submissions-{self.username}.pickle")