734 lines
25 KiB
Python
734 lines
25 KiB
Python
|
#!/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()
|