mod update bot

gistfile1.txt

import logging
import sys
import time

import httpx

logging.basicConfig(level=logging.DEBUG)


class Scraper(object):
    """
    Main scraper object
    """
    mod_portal = "https://mods.factorio.com/api/mods"

    def __init__(self, channel_id: int, bot_token: str):
        self.channel_id = channel_id
        self.bot_token = bot_token
        self.mods = {}

        self.populating = True

        self.channel_url = "https://discordapp.com/api/v6/channels/{}/messages".format(channel_id)

    def scraper_loop(self):
        headers = {
            "user-agent": "factorio mod notifier/2.0",
        }
        params = {"page_size": "99999"}

        while True:
            if not self.populating:
                time.sleep(600)
                pass
            else:
                print("Populating all mods...")

            for try_ in range(0, 10):
                print("Scanning mods... (try {} out of 10)".format(try_ + 1))

                try:
                    page = httpx.get(self.mod_portal, headers=headers, params=params, timeout=30)
                except httpx.ReadTimeout:
                    print("Timed out.")
                    continue
                except ConnectionError:
                    print("Connection failed.")
                    continue
                else:
                    print("Successfully downloaded page.")

                if page.status_code != 200:
                    print("Mod portal returned code {}, not 200".format(page.status_code))
                    continue

                body = page.json()
                break
            else:
                print("Failed to download mod info in 10 tries.")
                continue

            # actual mod notifier loop
            # dupe checks
            already_processed = set()
            for mod in body["results"]:
                name = mod.get("name")
                if not name:
                    continue

                if name in already_processed:
                    continue

                already_processed.add(name)

                latest_release = mod.get("latest_release")
                if not latest_release:
                    print("Mod {} has no latest release?".format(name))
                    continue

                version = latest_release.get("version")
                if not version:
                    print("Mod {} has no version?".format(version))
                    continue

                # compare to the currrent mod dict
                # if it doesn't exist and we're not populating, send a new message
                # if it does exist, and the version is changed, send a new message
                if not self.populating:
                    if name not in self.mods:
                        print("New mod: ", name)
                        self.new_mod(mod)
                    elif self.mods[name] != version:
                        print("Updated mod: ", name)
                        self.updated_mod(mod)
                else:
                    print("Populated mod", name)

                self.mods[name] = version

            print("Processed {} mods.".format(len(already_processed)))

            # always make sure this is unset
            self.populating = False

    def send_discord(self, message: str):
        try:
            headers = {
                "authorization": "Bot {}".format(self.bot_token),
                "content-type": "application/json; charset=utf-8",
                "dnt": "1",
                "user-agent": "DiscordBot (private)"
            }
            body = {"content": message}

            httpx.post(self.channel_url, headers=headers, json=body)
            # avoid ratelimits, messages are 5/5
            time.sleep(1.1)
        except Exception as e:
            print("Caught exception", e)

    def new_mod(self, mod):
        """
        Sends a new mod message.
        """
        mod_name = mod["name"]
        mod_url = "https://mods.factorio.com/mods/{}/{}".format(mod["owner"], mod_name)
        mod_message = "**New mod detected:** {} by {} - <{}>".format(
            mod["title"], mod["owner"], mod_url
        )

        self.send_discord(mod_message)

    def updated_mod(self, mod):
        """
        Sends an updated mod message.
        """
        mod_name = mod["name"]
        mod_url = "https://mods.factorio.com/mods/{}/{}".format(mod["owner"], mod_name)
        msg = "**Updated mod:** {} (updated to version: {}); by {} - <{}>".format(
            mod["title"], mod["latest_release"]["version"],
            mod["owner"], mod_url
        )

        self.send_discord(msg)


if __name__ == "__main__":
    CHANNEL_ID = 390979829907980289
    TOKEN = sys.argv[1]

    scraper = Scraper(channel_id=CHANNEL_ID, bot_token=TOKEN)
    scraper.scraper_loop()