Add Factorio/mod_updater.py

From Github: https://github.com/pdemonaco/factorio-mod-updater
Check there for updates

Factorio/mod_updater.py

@@ -0,0 +1,733 @@
+#!/usr/bin/python3
+"""
+This module provides a simple method to manage updating and installing mods
+on a given factorio server.
+
+It is currently not intended to be imported and instead should be executed
+directly as a python script.
+"""
+import argparse
+from collections import OrderedDict
+from datetime import datetime
+from enum import Enum, auto
+import glob
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+
+# External URL processing library
+# http://docs.python-requests.org/en/master/user/quickstart/
+import requests
+
+
+def _validate_hash(checksum: str, target: str, bsize: int = 65536) -> bool:
+    """
+    Checks to see if the file specified by target matches the provided sha1
+    checksum.
+
+    Keyword Arguments:
+    checksum -- sha1 digest to be matched
+    target   -- path to the file which must be validated
+    """
+    hasher = hashlib.sha1()
+
+    with open(target, "rb") as target_fp:
+        block = target_fp.read(bsize)
+        while len(block) > 0:
+            hasher.update(block)
+            block = target_fp.read(bsize)
+
+    return hasher.hexdigest() == checksum
+
+
+def _version_match(installed: str, mod: str):
+    """Checks if factorio versions are compatible."""
+    if installed.startswith("1.") and mod == "0.18":
+        return True
+    return installed == mod
+
+
+class ModUpdater:
+    """
+    Internal class managing the current version and state of the mods on this
+    server.
+    """
+
+    MOD_VERSION_PATTERN = r"\d+[.]\d+[.]\d+"
+    MOD_FILE_PATTERN = "^(.*)_({version})[.]zip$".format(version=MOD_VERSION_PATTERN)
+
+    class Mode(Enum):
+        """Possible execution modes"""
+
+        LIST = auto()
+        UPDATE = auto()
+
+    def __init__(
+        self,
+        settings_path: str,
+        data_path: str,
+        mod_path: str,
+        fact_path: str,
+        creds: hash,
+        title_mode: bool,
+    ):
+        """
+        Initialize the updater class with all mandatory and optional arguments.
+
+        Keyword arguments:
+        settings_path -- absolute path to the server-settings.json file
+        mod_path      -- absolute path to the factorio mod directory
+        fact_ver      -- local factorio version
+        """
+        self.mod_server_url = "https://mods.factorio.com"
+        self.mod_path = mod_path
+        self.timestamp = datetime.utcnow()
+        self.title_mode = title_mode
+
+        # Get the credentials to download mods
+        if settings_path is not None:
+            self.settings = self._parse_settings(settings_path)
+        else:
+            self.settings = {}
+        if data_path is not None:
+            self.data = self._parse_settings(data_path)
+        else:
+            self.data = {}
+
+        # Parse username and token
+        if "username" in creds and creds["username"] is not None:
+            self.username = creds["username"]
+        elif "username" in self.settings:
+            self.username = self.settings["username"]
+        elif "service-username" in self.data:
+            self.username = self.data["service-username"]
+        else:
+            self.token = None
+
+        if "token" in creds and creds["token"] is not None:
+            self.token = creds["token"]
+        elif "token" in self.settings:
+            self.token = self.settings["token"]
+        elif "service-token" in self.data:
+            self.token = self.data["service-token"]
+        else:
+            self.token = None
+
+        # Ensure username and token were specified
+        if self.username is None or self.username == "":
+            errmsg = (
+                "error: username not specified in "
+                + "server-settings.json, player-data.json, or cli!"
+            )
+            print(errmsg, file=sys.stderr)
+            sys.exit(1)
+
+        if self.token is None or self.token == "":
+            errmsg = (
+                "error: token not specified in "
+                + "server-settings.json, player-data.json, or cli!"
+            )
+            print(errmsg, file=sys.stderr)
+            sys.exit(1)
+
+        # Begin processing
+        self._determine_version(fact_path)
+        self._parse_mod_list()
+        self._retrieve_metadata()
+        self._determine_max_name_lengths()
+        if self.title_mode:
+            self.mods = OrderedDict(
+                sorted(self.mods.items(), key=lambda mod: mod[1]["title"])
+            )
+        else:
+            self.mods = OrderedDict(sorted(self.mods.items()))
+
+    def _determine_version(self, fact_path: str):
+        """Determine the local factorio version"""
+        if not os.path.exists(fact_path):
+            errmsg = "error: factorio binary '{fpath_path}' does not exist!"
+            print(errmsg, file=sys.stderr)
+            sys.exit(1)
+
+        try:
+            output = subprocess.check_output(
+                [fact_path, "--version"], universal_newlines=True
+            )
+            ver_re = re.compile(r"Version: (\d+)[.](\d+)[.](\d+) .*\n", re.RegexFlag.M)
+            match = ver_re.match(output)
+            if match:
+                version = {}
+                version["major"] = match.group(1)
+                version["minor"] = match.group(2)
+                version["patch"] = match.group(3)
+                version["release"] = "{}.{}".format(version["major"], version["minor"])
+                self.fact_version = version
+            else:
+                errmsg = "Unable to parse version from:\n{output}".format(output=output)
+                print(errmsg, file=sys.stderr)
+                sys.exit("1")
+
+        except subprocess.CalledProcessError as error:
+            errmsg = ("error: failed to run  '{fpath} --version': " "{errstr}").format(
+                fpath=fact_path, errstr=error.stderr
+            )
+            print(errmsg, file=sys.stderr)
+            sys.exit(1)
+
+        print(
+            "Factorio Release: {release}\n".format(release=self.fact_version["release"])
+        )
+
+    @staticmethod
+    def _parse_settings(config_path: str):
+        """Process the specified server-settings.json or player-data.json file."""
+        try:
+            with open(config_path, "r") as config_fp:
+                return json.load(config_fp)
+        except IOError as error:
+            errmsg = ("error: failed to open file '{fname}': " "{errstr}").format(
+                fname=config_path, errstr=error.strerror
+            )
+            print(errmsg, file=sys.stderr)
+            sys.exit(1)
+        except json.JSONDecodeError as error:
+            errmsg = ("error: failed to parse json file '{fname}': " "{errstr}").format(
+                fname=config_path, errstr=error.msg
+            )
+            print(errmsg, file=sys.stderr)
+            sys.exit(1)
+
+    def _retrieve_metadata(self):
+        """
+        Pull the latest metadata for each mod from the factorio server
+        See https://wiki.factorio.com/Mod_portal_API for details
+        """
+        print("Retrieving metadata", end="")
+        for mod, data in self.mods.items():
+            self._retrieve_mod_metadata(mod)
+            print(".", end="", flush=True)
+        print("complete!\n")
+
+        # Add missing dependencies to the overall list
+        while True:
+            missing_mods = []
+            for mod, data in self.mods.items():
+                if "missing_deps" in data:
+                    missing_mods.extend(data["missing_deps"])
+
+            unique_missing = set(missing_mods)
+            for mod in self.mods.keys():
+                if mod in unique_missing:
+                    unique_missing.remove(mod)
+            if len(unique_missing) == 0:
+                break
+            for mod in unique_missing:
+                entry = {}
+                entry["enabled"] = True
+                entry["installed"] = False
+                self.mods[mod] = entry
+                self._retrieve_mod_metadata(mod)
+                print("Info: adding missing dependency {dep}".format(dep=mod))
+
+        for mod, data in self.mods.items():
+            if "metadata" not in data:
+                warnmsg = (
+                    "Warning: Unable to retrieve metadata for"
+                    " {mod}, skipped!".format(mod=mod)
+                )
+                print(warnmsg)
+
+    def _retrieve_mod_metadata(self, mod: str):
+        """
+        Attempts to retrieve the metadata for the target mod. If found, the
+        data object is updated with the 'metadata' key and the 'latest' keys.
+        """
+        data = self.mods[mod]
+        mod_url = self.mod_server_url + "/api/mods/" + mod + "/full"
+        with requests.get(mod_url) as req:
+            if req.status_code == 200:
+                data["metadata"] = req.json()
+
+        if "metadata" in data:
+            # Find the latest release for this version of Factorio
+            matching_releases = []
+            for rel in data["metadata"]["releases"]:
+                rel_ver = rel["info_json"]["factorio_version"]
+                if _version_match(installed=self.fact_version["release"], mod=rel_ver):
+                    matching_releases.append(rel)
+
+            if len(matching_releases) > 0:
+                data["latest"] = matching_releases[-1]
+
+            # Add title key
+            data["title"] = data["metadata"]["title"]
+
+            # Mark whether it's deprecated
+            data["deprecated"] = data["metadata"].get("deprecated", False)
+        else:
+            data["title"] = mod
+
+            # Assume not deprecated if we can't find it
+            data["deprecated"] = False
+
+        if "latest" in data:
+            self._resolve_dependencies(mod)
+
+    def _resolve_dependencies(self, mod: str):
+        """
+        Processes the dependency list for this mod and returns an array
+        listing those which are not currently enabled. Note that this skips
+        exclusions and optional dependencies. (! and ?)
+        """
+        data = self.mods[mod]
+        if "latest" in data:
+            data["missing_deps"] = []
+            data["dependencies"] = {}
+            dependencies = data["latest"]["info_json"]["dependencies"]
+            # Preparation for future explicit version matching
+            dep_pattern = re.compile(r"^([\w -]+) ([<=>][=])? (\d+[.]\d+[.]\d+)$")
+            for dep_entry in dependencies:
+                match = dep_pattern.fullmatch(dep_entry)
+                if match:
+                    dep = {}
+                    dep_name = match.group(1)
+                    if dep_name == "base":
+                        continue
+                    dep["argument"] = match.group(2)
+                    dep["version"] = match.group(3)
+                    data["dependencies"][match.group(1)] = dep
+
+            for dep_name in data["dependencies"].keys():
+                if dep_name not in self.mods:
+                    data["missing_deps"].append(dep_name)
+
+    def _parse_mod_list(self):
+        """Process the mod-list.json within mod_path."""
+        mod_list_path = os.path.join(self.mod_path, "mod-list.json")
+        try:
+            settings_fp = open(mod_list_path, "r")
+            mod_json = json.load(settings_fp)
+            self.mods = {}
+            if "mods" in mod_json:
+                for mod in mod_json["mods"]:
+                    entry = {}
+                    entry["enabled"] = mod["enabled"]
+                    self.mods[mod["name"]] = entry
+            else:
+                print(
+                    "Invalid mod-list.json file \
+                      '{path}'!".format(
+                        path=mod_list_path
+                    ),
+                    file=sys.stderr,
+                )
+                sys.exit(1)
+
+            # Remove the 'base' mod as it's not relevant to this process
+            if "base" in self.mods:
+                del self.mods["base"]
+        except IOError as error:
+            errmsg = ("error: failed to open file '{fname}': " "{errstr}").format(
+                fname=mod_list_path, errstr=error.strerror
+            )
+            print(errmsg, file=sys.stderr)
+            sys.exit(1)
+        except json.JSONDecodeError as error:
+            errmsg = ("error: failed to parse json file '{fname}': " "{errstr}").format(
+                fname=mod_list_path, errstr=error.msg
+            )
+            print(errmsg, file=sys.stderr)
+            sys.exit(1)
+
+        # Collect the installed state & versions
+        self.mod_files = glob.glob("{mod_path}/*.zip".format(mod_path=self.mod_path))
+        installed_mods = {}
+        mod_pattern = re.compile(self.MOD_FILE_PATTERN)
+        for entry in self.mod_files:
+            basename = os.path.basename(entry)
+            match = mod_pattern.fullmatch(basename)
+            if match:
+                installed_mods[match.group(1)] = match.group(2)
+
+        for mod, data in self.mods.items():
+            if mod in installed_mods:
+                data["installed"] = True
+                data["version"] = installed_mods[mod]
+            else:
+                data["installed"] = False
+
+    def _update_mod_list(self):
+        """
+        Generates an updated mod-list.json file which takes into account any
+        newly added dependencies.
+        """
+        # Build the simplified object for json output
+        mod_list_output = {}
+        mod_list_output["mods"] = []
+        for mod, data in self.mods.items():
+            mod_entry = {}
+            mod_entry["name"] = mod
+            mod_entry["enabled"] = data["enabled"]
+            mod_list_output["mods"].append(mod_entry)
+
+        # Rename the old mod-list file with a timestamp
+        mod_list_path = os.path.join(self.mod_path, "mod-list.json")
+        mod_list_backup_path = os.path.join(
+            self.mod_path,
+            "mod-list.{timestamp}.json".format(
+                timestamp=self.timestamp.strftime("%Y-%m-%d_%H%M.%S")
+            ),
+        )
+        try:
+            os.rename(src=mod_list_path, dst=mod_list_backup_path)
+        except IOError as error:
+            errmsg = (
+                "error: failed to rename file '{s}' to '{d}': " "{errstr}"
+            ).format(s=mod_list_path, d=mod_list_backup_path, errstr=error.strerror)
+            print(errmsg, file=sys.stderr)
+            sys.exit(1)
+
+        # Store the current mod list
+        try:
+            mod_list_fp = open(mod_list_path, "w")
+            mod_list_fp.write(json.dumps(mod_list_output, indent=2, sort_keys=True))
+        except IOError as error:
+            errmsg = (
+                "error: failed to store updated mod list file '{s}': " "{errstr}"
+            ).format(s=mod_list_path, errstr=error.strerror)
+            print(errmsg, file=sys.stderr)
+            sys.exit(1)
+
+    def _determine_max_name_lengths(self):
+        """Returns the length of the longest mod name"""
+        max_mod_len = 0
+        max_cver_len = 0
+        max_lver_len = 0
+        for mod, data in self.mods.items():
+            mod_len = len(data["title"]) if self.title_mode else len(mod)
+            max_mod_len = mod_len if mod_len > max_mod_len else max_mod_len
+            cver_len = len(data["version"]) if data["installed"] else len("Version")
+            max_cver_len = cver_len if cver_len > max_cver_len else max_cver_len
+            lver_len = (
+                len(data["latest"]["version"]) if "latest" in data else len("Version")
+            )
+            max_lver_len = lver_len if lver_len > max_lver_len else max_lver_len
+
+        self.max_mod_len = max_mod_len
+        self.max_cver_len = max_cver_len
+        self.max_lver_len = max_lver_len
+        self.max_ver_len = max_lver_len if max_lver_len > max_cver_len else max_cver_len
+
+    def list(self):
+        """Lists the mods installed on this server."""
+        # Find the longest mod name
+
+        print(
+            "{:<{width}}\tenabled\tinstalled\tcurrent_v\tlatest_v".format(
+                "mod_name", width=self.max_mod_len
+            )
+        )
+        for mod, data in self.mods.items():
+            print(
+                "{:<{width}}\t{enbld}\t{inst}\t\t{cver}\t\t{lver}".format(
+                    mod,
+                    enbld=str(data["enabled"]),
+                    inst=str(data["installed"]),
+                    cver=data["version"] if data["installed"] else "N/A",
+                    lver=data["latest"]["version"] if "latest" in data else "N/A",
+                    width=self.max_mod_len,
+                )
+            )
+
+    def override_credentials(self, username: str, token: str):
+        """Replaces the values provided in server-settings.json or player-data.json"""
+        if username is not None:
+            self.username = username
+        if token is not None:
+            self.token = token
+
+    def _print_mod_message(
+        self, mod: str, version: str, action: str, result: str, message: str, data: hash
+    ):
+        """
+        Prints a mod status message using the provided parameters.
+        """
+        if data is not None:
+            title = data["title"] if self.title_mode else mod
+        else:
+            title = mod
+
+        output_string = (
+            "{title:<{mwidth}}\t{version:<{vwidth}}"
+            "\t{action:<10}\t{result:<10}\t{message}"
+        ).format(
+            title=title,
+            version=version,
+            action=action,
+            result=result,
+            message=message,
+            vwidth=self.max_ver_len,
+            mwidth=self.max_mod_len,
+        )
+        print(output_string)
+
+    def update(self):
+        """
+        Updates all mods currently installed on this server to the latest
+        release
+        """
+        self._print_mod_message("Mod", "Version", "Action", "Result", "Message", None)
+
+        for mod, data in self.mods.items():
+            version = data["version"] if data["installed"] else "N/A"
+            if "metadata" not in data:
+                self._print_mod_message(
+                    mod=mod,
+                    version=version,
+                    action="Skip",
+                    result="N/A",
+                    message="Missing metadata, skipping update!",
+                    data=data,
+                )
+                continue
+            if "latest" not in data:
+                message = (
+                    "No release found for factorio '{version}', skipping update!"
+                ).format(version=self.fact_version["release"])
+                self._print_mod_message(
+                    mod=mod,
+                    version=version,
+                    action="Skip",
+                    result="N/A",
+                    message=message,
+                    data=data,
+                )
+                continue
+
+            self._prune_old_releases(mod)
+            self._download_latest_release(mod)
+
+        # Update the mod list file
+        self._update_mod_list()
+
+    def _prune_old_releases(self, mod: str):
+        """
+        Deletes any locally installed versions older than the latest release.
+
+        Keyword Arguments:
+        mod -- name of the target to update
+        """
+        data = self.mods[mod]
+        latest_version = data["latest"]["version"]
+
+        # Declare the patterns
+        mod_pattern = re.compile(
+            "^{mod}_({ver})[.]zip$".format(mod=mod, ver=self.MOD_VERSION_PATTERN)
+        )
+        version_pattern = re.compile(
+            "^{mod}_{ver}[.]zip$".format(mod=mod, ver=latest_version)
+        )
+
+        # Build the parse list
+        basenames = [os.path.basename(x) for x in self.mod_files]
+        inst_rels = [x for x in basenames if mod_pattern.fullmatch(x)]
+        for rel in inst_rels:
+            if version_pattern.fullmatch(rel):
+                continue
+
+            match = mod_pattern.fullmatch(rel)
+            if match:
+                rel_ver = match.group(1)
+            else:
+                rel_ver = "TBD"
+
+            rel_path = os.path.join(self.mod_path, rel)
+            try:
+                os.remove(rel_path)
+                result = "Success"
+                message = ""
+            except OSError as error:
+                message = ("error: failed to remove '{fname}': " "{errstr}").format(
+                    fname=rel_path, errstr=error.strerror
+                )
+                result = "Failure"
+
+            self._print_mod_message(
+                mod=mod,
+                version=rel_ver,
+                action="Remove",
+                result=result,
+                message=message,
+                data=data,
+            )
+
+    def _download_latest_release(self, mod: str):
+        """
+        Retrieves the latest version of the specified mod compatible with the
+        factorio release present on this server.
+
+        Keyword Arguments:
+        mod -- name of the target to update
+        """
+        data = self.mods[mod]
+        latest = data["latest"]
+        target = os.path.join(self.mod_path, latest["file_name"])
+
+        validate = download = False
+
+        v_cur = data["version"] if "version" in data else "N/A"
+        v_new = latest["version"]
+        if data["installed"]:
+            if v_new == v_cur:
+                validate = True
+            else:
+                message = "Updating from '{v_cur}'".format(v_cur=v_cur)
+                download = True
+        else:
+            message = "Downloading initial release '{v_new}'".format(v_new=v_new)
+            download = True
+
+        if validate:
+            if _validate_hash(latest["sha1"], target):
+                result = "Success"
+                message = "Deprecated mod" if data["deprecated"] else ""
+            else:
+                result = "Failure"
+                download = True
+                message = "Validation failed, downloading again"
+            self._print_mod_message(
+                mod=mod,
+                version=v_cur,
+                action="Validate",
+                result=result,
+                message=message,
+                data=data,
+            )
+
+        if download:
+            creds = {"username": self.username, "token": self.token}
+            dl_url = self.mod_server_url + latest["download_url"]
+            with requests.get(dl_url, params=creds, stream=True) as req:
+                if req.status_code == 200:
+                    with open(target, "wb") as target_file:
+                        shutil.copyfileobj(req.raw, target_file)
+                        target_file.flush()
+                    if _validate_hash(latest["sha1"], target):
+                        result = "Success"
+                    else:
+                        result = "Failure"
+                        message = "Download did not match checksum!"
+                elif req.status_code == 403:
+                    message = (
+                        "Failed to download, credentials not accepted. "
+                        + "Check your username/token"
+                    )
+                    result = "Failure"
+                else:
+                    message = "Unable to retrieve, status code: " + str(req.status_code)
+                    result = "Failure"
+
+            self._print_mod_message(
+                mod=mod,
+                version=v_new,
+                action="Download",
+                result=result,
+                message=message,
+                data=data,
+            )
+
+
+if __name__ == "__main__":
+    DESC_TEXT = "Updates mods for a target factorio installation"
+    PARSER = argparse.ArgumentParser(description=DESC_TEXT)
+    # Username
+    PARSER.add_argument(
+        "-u",
+        "--username",
+        dest="username",
+        help="factorio.com username overriding server-settings.json/player-data.json",
+    )
+    # Token
+    PARSER.add_argument(
+        "-t",
+        "--token",
+        dest="token",
+        help="factorio.com API token overriding server-settings.json/player-data.json",
+    )
+    # Title format
+    PARSER.add_argument(
+        "--print-titles",
+        dest="title_mode",
+        default=False,
+        action="store_true",
+        help="When true, print the mod title instead of the api name",
+    )
+    # Server Settings
+    PARSER.add_argument(
+        "-s",
+        "--server-settings",
+        dest="settings_path",
+        required=False,
+        help=(
+            "Absolute path to the server-settings.json file "
+            + "(overrides player-data.json)"
+        ),
+    )
+    # Player Data
+    PARSER.add_argument(
+        "-d",
+        "--player-data",
+        dest="data_path",
+        required=False,
+        help="Absolute path to the player-data.json file",
+    )
+    # Factorio mod directory
+    PARSER.add_argument(
+        "-m",
+        "--mod-directory",
+        dest="mod_path",
+        required=True,
+        help="Absolute path to the mod directory",
+    )
+    # Factorio binary absolute path
+    PARSER.add_argument(
+        "--fact-path",
+        dest="fact_path",
+        required=True,
+        help="Absolute path to the factorio binary",
+    )
+    # Possible Execution modes
+    MODE_GROUP = PARSER.add_mutually_exclusive_group(required=True)
+    MODE_GROUP.add_argument(
+        "--list",
+        dest="mode",
+        action="store_const",
+        const=ModUpdater.Mode.LIST,
+        help="List the currently installed mods with versions",
+    )
+    MODE_GROUP.add_argument(
+        "--update",
+        dest="mode",
+        action="store_const",
+        const=ModUpdater.Mode.UPDATE,
+        help="Update all mods to their latest release",
+    )
+
+    ARGS = PARSER.parse_args()
+    UPDATER = ModUpdater(
+        settings_path=ARGS.settings_path,
+        data_path=ARGS.data_path,
+        mod_path=ARGS.mod_path,
+        fact_path=ARGS.fact_path,
+        creds={"username": ARGS.username, "token": ARGS.token},
+        title_mode=ARGS.title_mode,
+    )
+
+    if ARGS.mode == ModUpdater.Mode.LIST:
+        UPDATER.list()
+    elif ARGS.mode == ModUpdater.Mode.UPDATE:
+        UPDATER.update()