|
import random
|
|
import re
|
|
import time
|
|
|
|
from ragnar import log
|
|
from ragnar.api import Api
|
|
from ragnar.cache import method_cache
|
|
|
|
|
|
class Bot:
|
|
|
|
def __init__(self, username, password):
|
|
self.username = username
|
|
self.password = password
|
|
self.name = self.username.split("@")[0]
|
|
self.rant_history = []
|
|
self.amount_of_rants_to_check = 30
|
|
self.sus_users = ["Buffon"]
|
|
self.names = {
|
|
"no-spam": "Anna",
|
|
"no-spam1": "Ira",
|
|
"no-spam2": "Katya",
|
|
"no-spam3": "Nastya",
|
|
"no-spam4": "Vira",
|
|
}
|
|
self.name = self.names.get(self.name, self.username.split("@")[0])
|
|
self.mark_text = f"I am {self.name} and downvoted this post because I consider it spam. Your message will be removed from this community site due too much downvotes. See my profile for more information. Read my source code mentioned on my profile to see what you did wrong. Should be no problem for a developer.\n\nHave a nice day!\n\n\nIf your post is not spam, please mention @retoor."
|
|
self.auth = None
|
|
self.triggers = [
|
|
"$",
|
|
"crypto",
|
|
"hacker",
|
|
"recovery",
|
|
{"regex": r"\([+,(,0-9,),-]{7,}"},
|
|
{"regex": r"([\w\d]\.[\w\d]{2,5}\/[\w\d]+|http|www)"},
|
|
"money",
|
|
"landscape",
|
|
"cyber",
|
|
"recover",
|
|
"trust",
|
|
"bitcoin",
|
|
"wizard",
|
|
"diamond",
|
|
"carat",
|
|
"carats",
|
|
"leading",
|
|
"helping",
|
|
"unqiue",
|
|
"@ragnar",
|
|
"brides",
|
|
"singles",
|
|
"course",
|
|
"dating",
|
|
]
|
|
self.api = Api(username=self.username, password=self.password)
|
|
|
|
def rsleepii(self):
|
|
time.sleep(random.randint(1, 300))
|
|
|
|
@method_cache
|
|
def login(self):
|
|
self.rsleepii()
|
|
self.auth = self.api.login()
|
|
if not self.auth:
|
|
log.error(f"Authentication for {self.username} failed.")
|
|
raise Exception("Login error")
|
|
log.info(f"Authentication succesful for {self.username}.")
|
|
|
|
def clean_rant_text(self, rant_text):
|
|
return rant_text.replace(" ", "").lower()
|
|
|
|
@method_cache
|
|
def is_sus_content(self, content):
|
|
clean_text = self.clean_rant_text(content)
|
|
for trigger in self.triggers:
|
|
if type(trigger) == dict:
|
|
if trigger.get("regex"):
|
|
regex = trigger["regex"]
|
|
if re.search(regex, clean_text):
|
|
log.info(f"Regex trigger {regex} matched!")
|
|
return True
|
|
elif trigger in clean_text:
|
|
log.info(f"Trigger {trigger} matched!")
|
|
return True
|
|
return False
|
|
|
|
@method_cache
|
|
def is_sus_rant(self, rant_id, rant_text):
|
|
return self.is_sus_content(rant_text)
|
|
|
|
def is_flagged_as_sus(self, rant_id, num_comments):
|
|
if not num_comments:
|
|
return False
|
|
rant = self.api.get_rant(rant_id)
|
|
for comment in rant.get("comments", []):
|
|
if self.names.get(self.username, "") in comment.get("body", ""):
|
|
return True
|
|
return False
|
|
|
|
@method_cache
|
|
def is_user_sus(self, username):
|
|
if username in self.sus_users:
|
|
return True
|
|
user_id = self.api.get_user_id(username)
|
|
profile = self.api.get_profile(user_id)
|
|
score = profile["score"]
|
|
if score < 5:
|
|
log.warning(f"User {username} is sus with his score of only {score}.")
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def is_comments_sus(self, rant_id):
|
|
log.info(f"Checking if comments are sus of rant {rant_id}.")
|
|
rant = self.api.get_rant(rant_id)
|
|
for comment in rant.get("comments", []):
|
|
print("Checking if sus comment: ", comment["body"])
|
|
if "@ragnar" in comment.get("body", "").lower():
|
|
print("Ragnar is mentioned, so flagging as sus comment.")
|
|
return True
|
|
if comment["user_score"] >= 5:
|
|
print("User has reputation >= 5 so not sus.")
|
|
continue
|
|
if self.is_sus_content(comment.get("body", "")):
|
|
return True
|
|
return False
|
|
|
|
def mark_as_sus(self, rant):
|
|
self.api.post_comment(rant["id"], self.mark_text)
|
|
|
|
def fight(self):
|
|
self.rsleepii()
|
|
rants = self.api.get_rants("recent", self.amount_of_rants_to_check, 0)
|
|
for rant in rants:
|
|
if rant["id"] in self.rant_history:
|
|
log.debug("{}: Already checked rant {}.".format(self.name, rant["id"]))
|
|
continue
|
|
|
|
if not self.is_user_sus(rant["user_username"]):
|
|
log.info(
|
|
"{}: User {} is trusted.".format(self.name, rant["user_username"])
|
|
)
|
|
self.rant_history.append(rant["id"])
|
|
continue
|
|
elif self.is_comments_sus(rant["id"]):
|
|
|
|
log.info("Comments of rant are sus + user sus. Will flag as spam.")
|
|
elif rant["user_username"] in self.sus_users:
|
|
pass
|
|
elif not self.is_sus_rant(rant["id"], rant["text"]):
|
|
log.info(
|
|
"{}: Rant by {} is not sus.".format(
|
|
self.name, rant["user_username"]
|
|
)
|
|
)
|
|
continue
|
|
|
|
log.warning(
|
|
"{}: Rant is not {} flagged as sus yet.".format(
|
|
self.name, rant["user_username"]
|
|
)
|
|
)
|
|
log.warning(
|
|
"{}: Flagging rant by {} as sus.".format(
|
|
self.name, rant["user_username"]
|
|
)
|
|
)
|
|
self.mark_as_sus(rant)
|
|
self.down_vote_rant(rant)
|
|
self.rant_history.append(rant["id"])
|
|
|
|
def down_vote_rant(self, rant):
|
|
log.warning("Downvoting rant by {}.".format(rant["user_username"]))
|
|
log.debug("Vote result: ".format(self.api.post_rant_vote(rant["id"], -1)))
|