This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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'. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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]) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
try: | |
... | |
except Exception as ex: | |
error_message = str(ex) | |
self._retry_rate_limited_failure(error_message, self.work_on_subreddit, subreddit, **generator_kwargs) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# main.py | |
import json | |
import logging | |
import os | |
from pathlib import Path | |
from bot import RedditBot | |
logging.basicConfig(handlers=[logging.StreamHandler()], | |
level=logging.INFO, |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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"] |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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") |