Skip to content

Instantly share code, notes, and snippets.

@binarybrat

binarybrat/streamingprocesses.py

Last active August 10, 2020 13:35
Show Gist options
  • Save binarybrat/2092cf49a27f1541ec0668e7154a6120 to your computer and use it in GitHub Desktop.
Save binarybrat/2092cf49a27f1541ec0668e7154a6120 to your computer and use it in GitHub Desktop.
Download ZIP
Current streaming processes setup - adding logger
Raw
streamingprocesses.py
import time
import multiprocessing
import time
from datetime import datetime
from ago import human
from peewee import IntegrityError
from loggingTesting import *
from models import init_db
from models.commentmodel import Comment, updateComment, delUpdateComment, add_comment, commentDone
from models.detailsmodel import Details, DetailsPending
from models.detailsmodel import DetailsHelper
from models.detailsmodel import get_details_approvalID
from models.messagemodel import SewingMessages, add_sewmsg
from models.modlogmodel import Modlog, most_actions, add_modlog
from models.patternmodel import Pattern, PendingPatternPosts, ppp_done, add_ppp
from models.postmodel import Posts, add_post, postDone, update_post
from models.postmodel import delUpdate_post, unflaired_canFlair
from models.statstrafficmodel_ import TotalsTesting
from utils import detailpendinglogin
from utils.globalvars import isdeleted, has_pattern, hardcore_removed, formatdt
from utils.globalvars import shortenRedditURL
from utils.randomhelpershit import FLAIRCSSTOCHECK, modpost, flair_post
def detailsComment(comment, wordmin=30):
if not isdeleted(comment) and not comment.removed and not comment.spam and not hardcore_removed(comment) and \
comment.is_root and comment.is_submitter and not comment.submission.spam and \
not hardcore_removed(comment.submission):
return True if (len(comment.body.split()) >= wordmin or has_pattern(comment.body)) else False
class StreamingStuff:
def __init__(self, wordmin=30, scoremin=50, timeback=24, doPatterns=False, nsfwChecks=False):
self.wordmin = wordmin
self.scoremin = scoremin
self.timeback = timeback
self.doPatterns = doPatterns
self.nsftChecks = nsfwChecks
self.reddit = detailpendinglogin.login()
self.sewsub = self.reddit.subreddit('sewing')
self.START_TIME = time.time()
self.HOW_AGO = self.START_TIME - (3600 * 24)
self.db = init_db()
self.db.connect()
self.db.create_tables([Comment, Posts, Details, DetailsPending, Modlog, SewingMessages, PendingPatternPosts, Pattern])
self.db.close()
self.log = Log()
self.startTime = datetime.now()
def post_stream(self):
self.log.LightBlue("Starting Post Stream ~ {}".format(datetime.now().strftime(formatdt)))
while True:
for post in self.sewsub.stream.submissions(skip_existing=True):
self.db.connect()
self.log.LightBlue(
">> Incoming post by u/{} - {}\n > {}".format(post.author.name if post.author else "[deleted]",
post.shortlink, post.title))
if not postDone(post) and not isdeleted(post):
add_post(post)
self.log.Intro(" > Added post to the database")
# try:
# TotalsTesting.create(
# datecreated=datetime.utcfromtimestamp(post.created_utc).date(),
# idstr=post.fullname
# )
# self.log.Intro(" > Added post to TotalsTesting DB")
# except IntegrityError:
# pass
else:
if isdeleted(post):
delUpdate_post(post)
self.log.Intro(" > Updating Deleted Post: {}\n > {}".format(post.shortlink, post.title))
else:
update_post(post)
self.log.Intro(" > Updating Post by u/{}".format(post.author.name if post.author else "[deleted]"))
# Approve weird recurring posts that AuoMod does not approve
if post.approved and post.approved_by is not None and not isdeleted(
post) and post.author.name in ['AutoModerator',
'sewingmodthings']:
post.mod.approve()
self.log.Intro(
" > Approved a weekly thread by AutoModerator: {}\n{}".format(post.shortlink,
post.title))
# Checks author account for over-18 and age to determine if should filter - half assed spambot protection
if self.nsftChecks and post.author.subreddit['over_18'] and (time.time() - post.author.created_utc <= 2629743):
post.mod.remove()
post.report("NSFW user with account under 30 days old - please check and take action")
self.log.Intro(" > NSFW user account under a month old - removed and sending report")
continue
# Adds flair to what looks to be a project type post that is unflaired
if post.link_flair_text is None and unflaired_canFlair(post):
flair_post(post, 'FO', False)
self.log.Intro(" > u/sewingmodthings gave FO flair to unflaired post")
# If pattern found in title add it to the pending patterns DB
if post.link_flair_text is not None and post.link_flair_css_class in FLAIRCSSTOCHECK and \
has_pattern(post.title) and not ppp_done(post):
try:
add_ppp(post)
self.log.Intro(" > Pattern found in posted - added to pending posts db")
except IntegrityError:
pass
self.db.close()
def comment_stream(self):
self.log.Magenta("Starting Comment Stream ~ {}".format(datetime.now().strftime(formatdt)))
while True:
for comment in self.sewsub.stream.comments(skip_existing=True):
self.db.connect()
self.log.Magenta(
"Incomming comment by u/{} - {}".format(comment.author.name if comment.author else "[deleted]",
comment.fullname))
if not commentDone(comment.fullname) and not isdeleted(comment):
add_comment(comment)
self.log.Intro(" > NEW Comment added to database")
# try:
# TotalsTesting.create(datecreated=datetime.utcfromtimestamp(comment.created_utc).date(),
# idstr=comment.fullname)
# self.log.Intro(" > Comment added to TotalsTesting DB")
# except IntegrityError:
# pass
else:
if isdeleted(comment):
delUpdateComment(comment)
self.log.Intro(" > Comment deleted - updated database")
else:
updateComment(comment)
self.log.Intro(" > Comment updated in the database")
# Checks if project type post and whether or not comment passes as a details comment/processed details
if detailsComment(comment, wordmin=self.wordmin):
self.log.Intro(" > Details Comment Found")
dhelper = DetailsHelper(self.reddit, comment, verbose=True, overkill=True)
if not dhelper.already_in_db() and not dhelper.is_deleted:
dhelper.add_details()
self.log.Intro(" > Added details to the DB")
else:
if dhelper.oktoupdate:
self.log.Intro(" > Updated details comment in DB")
if dhelper.has_pattern and not ppp_done(dhelper.comment.submission):
if has_pattern(dhelper.comment.submission.title):
add_ppp(dhelper.submission)
self.log.Intro(" > Adding POST pending pattern post: {} - {}".format(dhelper.comment.fullname,
dhelper.submission.fullname))
else:
add_ppp(dhelper.submission, dhelper.comment.fullname, dhelper.comment.body)
self.log.Intro(
" > Adding COMMENT pending pattern post: {} - {}".format(dhelper.comment.fullname,
dhelper.submission.fullname))
# If post is either warned, or filtered process and send approval message
if dhelper.pending:
self.log.Intro(" > Found a details pending post with details comment!!!")
pending = DetailsPending.get_or_none(DetailsPending.idstr == dhelper.comment.submission.fullname)
if pending.filtered and dhelper.comment.removed and dhelper.comment.banned_by == 'sewingmodthings':
modpost(dhelper.submission.id, 'approve')
self.log.Intro(" > Re-Approved pending post for giving details")
if not dhelper.approval_msg_sent():
dhelper.send_details_approval()
apmsgid = get_details_approvalID(self.reddit, dhelper.submission)
self.log.Intro(" > Sent Details Approval Message: {}".format(apmsgid))
dhelper.update_pending_approved()
self.log.Intro(" > Updated pending details to approved: {} - {}".format(dhelper.comment.fullname,
dhelper.submission.fullname))
self.db.close()
def edited_stream(self):
self.log.Teal("Starting Edited Stream ~ {}".format(datetime.now().strftime(formatdt)))
while True:
for edited in self.sewsub.mod.stream.edited(skip_existing=True):
self.db.connect()
if edited.fullname.startswith('t1'):
normalprint = ">> Incoming Edited Comment by: {} - {}".format(
edited.author.name if edited.author else "[deleted]", edited.fullname)
detailsprint = ">> Incoming Edited DETAILS Comment by: {} - {}".format(
edited.author.name if edited.author else "[deleted]", edited.fullname)
self.log.Teal(detailsprint if detailsComment(edited) else normalprint)
if not commentDone(edited.fullname) and not isdeleted(edited):
add_comment(edited)
self.log.Intro(" > NEW Comment added to database")
try:
TotalsTesting.create(datecreated=datetime.utcfromtimestamp(edited.created_utc).date(),
idstr=edited.fullname)
except IntegrityError:
pass
else:
if isdeleted(edited):
delUpdateComment(edited)
self.log.Intro(" > Edited Comment deleted - updated database")
else:
updateComment(edited)
self.log.Intro(" > Edited Comment updated in the database")
if detailsComment(edited, wordmin=30):
self.log.Intro(" > Edited Details Comment Found")
dhelper = DetailsHelper(self.reddit, edited, verbose=True, overkill=True)
if not dhelper.already_in_db() and not dhelper.is_deleted:
dhelper.add_details()
self.log.Intro(" > Added details to the DB")
else:
if dhelper.oktoupdate:
self.log.Intro(" > Updated details comment in DB")
if dhelper.has_pattern and not dhelper.pattern_in_db():
if has_pattern(dhelper.comment.submission.title):
add_ppp(dhelper.submission)
self.log.Intro(
" > Adding POST pending pattern post: {} - {}".format(dhelper.comment.fullname,
dhelper.submission.fullname))
else:
add_ppp(dhelper.submission, dhelper.comment.fullname, dhelper.comment.body)
self.log.Intro(
" > Adding COMMENT pending pattern post: {} - {}".format(
dhelper.comment.fullname,
dhelper.submission.fullname))
if dhelper.pending:
pending = DetailsPending.get_or_none(
DetailsPending.idstr == dhelper.comment.submission.fullname)
if pending.filtered and dhelper.comment.removed and \
dhelper.comment.banned_by == 'sewingmodthings':
modpost(dhelper.submission.id, 'approve')
self.log.Intro(" > Re-Approved pending post for giving details")
if not dhelper.approval_msg_sent():
dhelper.send_details_approval()
apmsgid = get_details_approvalID(self.reddit, dhelper.submission)
self.log.Intro(" > Sent Details Approval Message: {}".format(apmsgid))
dhelper.update_pending_approved()
self.log.Intro(
" > Updated edited pending details to approved: {} - {}".format(
dhelper.comment.fullname,
dhelper.submission.fullname))
elif edited.fullname.startswith('t3'):
self.log.Teal(">> Incoming Edited Post by: u/{} - {}".format(edited.author.name if
edited.author else "[deleted]",
edited.fullname))
if not postDone(edited) and not isdeleted(edited):
add_post(edited)
self.log.Intro(" > NEW Post added to database FROM EDITED STREAM")
try:
TotalsTesting.create(
datecreated=datetime.utcfromtimestamp(edited.created_utc).date(),
idstr=edited.fullname
)
except IntegrityError:
pass
else:
if isdeleted(edited):
delUpdate_post(edited)
self.log.Intro(" > Edited post deleted - updated database")
else:
update_post(edited)
self.log.Intro(" > Updated edited post in database")
self.db.close()
def inbox_stream(self):
self.log.LightTeal("Starting Inbox Stream ~ {}".format(datetime.now().strftime(formatdt)))
while True:
for msg in self.reddit.inbox.stream(skip_existing=True):
self.log.LightTeal(">> Incoming Inbox item from: u/{} - {}".format(
msg.author.name if msg.author else "[deleted]",
msg.fullname))
self.db.connect()
try:
add_sewmsg(msg)
itemadded = "Message" if msg.fullname.startswith('t4_') else "Comment"
self.log.LightTeal(" > New {} Added: {} - {} - {}".format(itemadded, msg.author, msg.dest, msg.name))
except IntegrityError:
pass
self.db.close()
def modlog_stream(self):
self.log.Success("Starting Modlog Stream ~ {}".format(datetime.now().strftime(formatdt)))
while True:
for entry in self.sewsub.mod.stream.log(skip_existing=True):
self.db.connect()
if entry.action in most_actions:
self.log.Success(">>> Incoming Modlog Entry: {} to {} by u/{} | Mod: {}".format(
entry.action,
entry.target_fullname,
entry.target_author,
entry.mod))
try:
add_modlog(entry)
if entry.target_fullname is not None and entry.target_fullname.startswith('t3_') or \
entry.target_fullname.startswith('t1_'):
urlLink = shortenRedditURL('https://www.reddit.com' + entry.target_permalink)
self.log.Intro(" > {}: {}".format("Post" if entry.target_fullname.startswith('t3_')
else "Comment", urlLink))
if entry.target_fullname.startswith("t3_"):
self.log.Intro(" > {}".format(entry.target_title))
if entry.mod == 'AutoModerator':
self.log.Intro(" > Details: {}".format(entry.details))
except IntegrityError:
pass
if entry.target_fullname is not None and entry.target_fullname.startswith('t3_') and \
entry.action == 'editflair':
flair_edit_post = self.reddit.submission(entry.target_fullname.lstrip('t3_'))
if hasattr(flair_edit_post, 'link_flair_text'):
flairtext = flair_edit_post.link_flair_text if flair_edit_post.link_flair_text is not None else None
onthedb = Posts.get_or_none(Posts.idstr == flair_edit_post.id)
if onthedb.flairtext != flairtext:
update_post(flair_edit_post)
self.log.Intro(" > Updated flair edited post to proper flair in the DB")
self.db.close()
def processing(self):
subprocess1 = multiprocessing.Process(target=self.post_stream)
subprocess2 = multiprocessing.Process(target=self.comment_stream)
subprocess3 = multiprocessing.Process(target=self.edited_stream)
subprocess4 = multiprocessing.Process(target=self.inbox_stream)
subprocess5 = multiprocessing.Process(target=self.modlog_stream)
subprocess1.start()
subprocess2.start()
subprocess3.start()
subprocess4.start()
subprocess5.start()
if __name__ == '__main__':
startTime = time.time()
try:
StreamingStuff().processing()
except KeyboardInterrupt:
print("Start Time: {} - Ran for: {}".format(datetime.fromtimestamp(startTime).strftime(formatdt),
human(startTime)))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment