Initial commit

src/Ragnar.egg-info/PKG-INFO

@@ -0,0 +1,12 @@
+Metadata-Version: 2.1
+Name: Ragnar
+Version: 1.3.37
+Summary: Anti spam bot for dR
+Author: Retoor
+Author-email: retoor@molodetz.nl
+License: MIT
+Requires-Python: >=3.7
+Description-Content-Type: text/markdown
+Requires-Dist: aiohttp==3.10.10
+Requires-Dist: dataset==1.6.2
+Requires-Dist: requests==2.32.3

src/Ragnar.egg-info/SOURCES.txt

@@ -0,0 +1,14 @@
+pyproject.toml
+setup.cfg
+src/Ragnar.egg-info/PKG-INFO
+src/Ragnar.egg-info/SOURCES.txt
+src/Ragnar.egg-info/dependency_links.txt
+src/Ragnar.egg-info/entry_points.txt
+src/Ragnar.egg-info/requires.txt
+src/Ragnar.egg-info/top_level.txt
+src/ragnar/__init__.py
+src/ragnar/__main__.py
+src/ragnar/api.py
+src/ragnar/bot.py
+src/ragnar/cache.py
+src/ragnar/cli.py

src/Ragnar.egg-info/dependency_links.txt

@@ -0,0 +1 @@
+

src/Ragnar.egg-info/entry_points.txt

@@ -0,0 +1,2 @@
+[console_scripts]
+ragnar.run = ragnar.cli:run

src/Ragnar.egg-info/requires.txt

@@ -0,0 +1,3 @@
+aiohttp==3.10.10
+dataset==1.6.2
+requests==2.32.3

src/Ragnar.egg-info/top_level.txt

@@ -0,0 +1 @@
+ragnar

src/ragnar/api.py

@@ -0,0 +1,96 @@
+import requests, json
+
+from ragnar.cache import method_cache
+
+
+class Api:
+
+    base_url = "https://www.devrant.io/api/"
+
+    def __init__(self, username, password):
+        self.username = username
+        self.password = password
+        self.auth = None
+
+    def post_comment(self, rant_id, text):
+        response = requests.post(
+            self.base_url + "devrant/rants/" + str(rant_id) + "/comments",
+            data={
+                "app": 3,
+                "user_id": self.auth["user_id"],
+                "token_id": self.auth["token_id"],
+                "token_key": self.auth["token_key"],
+                "comment": text,
+            },
+        )
+        return response.json()
+
+    @method_cache
+    def login(self):
+        print("New login, cache miss?")
+        rawdata = requests.post(
+            self.base_url + "users/auth-token",
+            data={"username": self.username, "password": self.password, "app": 3},
+        )
+        rawdata = rawdata.json()
+        if not rawdata["success"]:
+            self.auth = None
+        else:
+            self.auth = {
+                "token_id": rawdata["auth_token"]["id"],
+                "token_key": rawdata["auth_token"]["key"],
+                "user_id": rawdata["auth_token"]["user_id"],
+            }
+        return self.auth
+
+    @method_cache
+    def get_profile(self, id_):
+        url = self.base_url + "users/" + str(id_)
+        params = {
+            "app": 3,
+        }
+        response = requests.get(url, params)
+        return json.loads(response.text)["profile"]
+
+    def get_search(self, term):
+        url = self.base_url + "devrant/search"
+        params = {"app": 3, "term": term}
+        response = requests.get(url, params, timeout=5)
+        obj = json.loads(response.text)
+        return obj
+
+    def post_rant_vote(self, id, vote):
+        response = requests.post(
+            self.base_url + "devrant/rants/" + str(id) + "/vote",
+            data={
+                "app": 3,
+                "user_id": self.auth["user_id"],
+                "token_id": self.auth["token_id"],
+                "token_key": self.auth["token_key"],
+                "vote": vote,
+                # "plat": 3,
+            },
+        )
+        return response.json()
+
+    def get_rant(self, rant_id):
+        url = self.base_url + "devrant/rants/" + str(rant_id)
+        params = {
+            "app": 3,
+        }
+        response = requests.get(url, params, timeout=5)
+        return json.loads(response.text)
+
+    def get_rants(self, sort, limit, skip):
+        url = self.base_url + "devrant/rants"
+        params = {"app": 3, "sort": sort, "limit": limit, "skip": skip}
+
+        response = requests.get(url, params, timeout=5)
+        return json.loads(response.text)["rants"]
+
+    @method_cache
+    def get_user_id(self, name):
+        url = self.base_url + "get-user-id"
+        params = {"app": 3, "username": name}
+        response = requests.get(url, params)
+        return json.loads(response.text).get("user_id", None)

src/ragnar/bot.py

@@ -0,0 +1,93 @@
+from ragnar.api import Api
+import time
+import random
+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]
+
+        names = {
+            "no-spam": "anna",
+            "no-spam1": "ira",
+            "no-spam2": "katya",
+            "no-spam3": "nastya",
+            "no-spam4": "vira",
+        }
+        self.name = names.get(self.name, "everyone")
+        self.mark_text = "You rant is flagged as spam by {}. Read bot source code to find out how to prevent this. Have a nice day! (Bot does not downvote yet, couldn't figure out what the downvote value should be. Upvote these bots btw so they can post a link. They're quite effective, they'll end spam.)".format(
+            self.name
+        )
+        self.auth = None
+        self.triggers = ["$", "crypto", "hacker", "recovery"]
+
+        self.api = Api(username=self.username, password=self.password)
+
+    def rsleepii(self):
+        time.sleep(random.randint(1, 3))
+
+    @method_cache
+    def login(self):
+        self.rsleepii()
+        self.auth = self.api.login()
+        if not self.auth:
+            print("Authentication for {} failed.".format(self.username))
+            raise Exception("Login error")
+        print("Authentication succesful for {}.".format(self.username))
+
+    @method_cache
+    def is_sus_rant(self, rant_id, rant_text):
+        clean_text = rant_text.replace(" ", "").lower()
+        for trigger in self.triggers:
+            if trigger in clean_text:
+                return True
+
+    def is_flagged_as_sus(self, rant_id, num_comments):
+        if not num_comments:
+            return False
+        self.rsleepii()
+        rant = self.api.get_rant(rant_id)
+        for comment in rant.get("comments", []):
+            if self.mark_text in comment.get("body", ""):
+                return True
+        return False
+
+    @method_cache
+    def is_user_sus(self, username):
+        user_id = self.api.get_user_id(username)
+        profile = self.api.get_profile(user_id)
+        score = profile["score"]
+        if score < 5:
+            print("User {} is sus with his score of only {}.".format(username, score))
+            return True
+        else:
+            return False
+
+    def mark_as_sus(self, rant):
+        self.rsleepii()
+        self.api.post_comment(rant["id"], self.mark_text)
+
+    def fight(self):
+        self.rsleepii()
+        rants = self.api.get_rants("recent", 5, 0)
+        for rant in rants:
+            if not self.is_user_sus(rant["user_username"]):
+                print("User {} is trusted.".format(rant["user_username"]))
+                continue
+            if not self.is_sus_rant(rant["id"], rant["text"]):
+                print("Rant by {} is not sus.".format(rant["user_username"]))
+                continue
+            if self.is_flagged_as_sus(rant["id"], rant.get("num_comments")):
+                continue
+            print("Rant is not {} flagged as sus yet.".format(rant["user_username"]))
+            print("Flagging rant by {} as sus.".format(rant["user_username"]))
+            self.mark_as_sus(rant)
+            self.down_vote_rant(rant)
+
+    def down_vote_rant(self, rant):
+        print("Downvoting rant by {}.".format(rant["user_username"]))
+        print(self.api.post_rant_vote(rant["id"], 4))

src/ragnar/cache.py

@@ -0,0 +1,14 @@
+# The functools lru_cache didn't work well on a class so
+# I had to create a custom cashing method. Which is fine.
+
+
+def method_cache(func):
+    cache = {}
+
+    def wrapper(*args, **kwargs):
+        key = (args, tuple(sorted(kwargs.items())))
+        if key not in cache:
+            cache[key] = func(*args, **kwargs)
+        return cache[key]
+
+    return wrapper

src/ragnar/cli.py

@@ -0,0 +1,42 @@
+import argparse
+from ragnar.bot import Bot
+import random
+import time
+from concurrent.futures import ThreadPoolExecutor as Executor
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description="Process username and password.")
+
+    parser.add_argument("-u", "--username", required=True, help="Your username")
+    parser.add_argument("-p", "--password", required=True, help="Your password")
+
+    return parser.parse_args()
+
+
+def bot_task(username, password):
+    time.sleep(random.randint(1, 20))
+    bot = Bot(username=username, password=password)
+    bot.login()
+    while True:
+        time.sleep(random.randint(1, 20))
+        try:
+            bot.fight()
+        except Exception as ex:
+            print(ex)
+
+
+def main():
+    args = parse_args()
+    with Executor(4) as executor:
+        for x in range(1, 5):
+            username = "no-spam{}@molodetz.nl".format(str(x))
+            password = args.password
+            time.sleep(1)
+            print("Starting bot {}.".format(username))
+            executor.submit(bot_task, username, password)
+        executor.shutdown(wait=True)
+
+
+def run():
+    main()