161 lines
4.5 KiB
Python
161 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
|
|
"""
|
|
A configurable Minecraft server (auto-)updater.
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
|
|
from socket import timeout as SocketTimeoutError
|
|
from subprocess import CalledProcessError, check_call
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import urlopen, urlretrieve
|
|
|
|
|
|
# Mojang's version manifest file.
|
|
VERSION_MANIFEST = 'https://launchermeta.mojang.com/mc/game/version_manifest.json'
|
|
|
|
|
|
def _sha1sum(file_path: str) -> str:
|
|
"""Return passed file SHA1 digest"""
|
|
sha1 = hashlib.sha1()
|
|
|
|
with open(file_path, 'rb') as file:
|
|
while True:
|
|
file_chunk = file.read(8192)
|
|
if not file_chunk:
|
|
break
|
|
|
|
sha1.update(file_chunk)
|
|
|
|
return sha1.hexdigest()
|
|
|
|
|
|
def _load_remote_json(url: str) -> dict:
|
|
"""Download and decode a WWW JSON object before returning it"""
|
|
try:
|
|
with urlopen(url) as http_request:
|
|
return json.load(http_request)
|
|
except (HTTPError, URLError, SocketTimeoutError) as error:
|
|
sys.exit(f"Couldn't fetch {url} : {error}")
|
|
|
|
|
|
def _get_latest_version_info(prefer_snapshot: bool) -> dict:
|
|
"""
|
|
From the Mojang's version manifest, fetch the latest snapshot/release identifier.
|
|
With it, find version's details and download its corresponding package meta-information.
|
|
From them, return the information related to server asset.
|
|
|
|
At this time, `sha1`, `size` & `url` keys are returned.
|
|
"""
|
|
version_manifest = _load_remote_json(VERSION_MANIFEST)
|
|
|
|
latest_version = version_manifest['latest'].get(
|
|
('snapshot' if prefer_snapshot else 'release')
|
|
)
|
|
|
|
for version in version_manifest['versions']:
|
|
if version['id'] == latest_version:
|
|
version_url = version['url']
|
|
break
|
|
else:
|
|
sys.exit("Internal error : Is Mojang's version manifest broken ?")
|
|
|
|
return _load_remote_json(version_url)['downloads']['server']
|
|
|
|
|
|
def _run_hook(command: str) -> bool:
|
|
"""Run passed command and check its exit status"""
|
|
try:
|
|
check_call(command, shell=True)
|
|
except CalledProcessError:
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
|
|
def main():
|
|
"""."""
|
|
parser = argparse.ArgumentParser(
|
|
description="A configurable Minecraft server (auto-)updater."
|
|
)
|
|
|
|
parser.add_argument(
|
|
'server_path',
|
|
help="Minecraft server binary path"
|
|
)
|
|
parser.add_argument(
|
|
'-i', '--initialization',
|
|
action='store_true',
|
|
help="Download the latest Minecraft server unconditionally"
|
|
)
|
|
parser.add_argument(
|
|
'-b', '--backup',
|
|
action='store_true',
|
|
help="Save a copy of the current server before updating it"
|
|
)
|
|
parser.add_argument(
|
|
'-s', '--snapshot',
|
|
action='store_true',
|
|
help="Allow updating to snapshots"
|
|
)
|
|
parser.add_argument(
|
|
'--pre-hook',
|
|
help="Hook command to run before updating"
|
|
)
|
|
parser.add_argument(
|
|
'--post-hook',
|
|
help="Hook command to run after updating"
|
|
)
|
|
parser.add_argument(
|
|
'-v', '--version',
|
|
action='version', version="%(prog)s : v1.0.3"
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not os.path.exists(args.server_path) and not args.initialization:
|
|
sys.exit("Server does not exist, pass `-i` to download it for the first time.")
|
|
|
|
current_hash = (not args.initialization and _sha1sum(args.server_path))
|
|
latest_version_info = _get_latest_version_info(args.snapshot)
|
|
|
|
if current_hash == latest_version_info['sha1']:
|
|
# Up to date, nothing to do.
|
|
sys.exit()
|
|
|
|
if args.pre_hook and not _run_hook(args.pre_hook):
|
|
sys.exit("Pre-hook command failed. Aborting.")
|
|
|
|
if args.backup and not args.initialization:
|
|
shutil.copy2(args.server_path, (args.server_path + '.bak'))
|
|
|
|
urlretrieve(latest_version_info['url'], args.server_path)
|
|
|
|
# A download verification is even possible !
|
|
status_code = 0
|
|
if _sha1sum(args.server_path) != latest_version_info['sha1']:
|
|
status_code = 1
|
|
print(
|
|
"Download verification failed, trying to roll-back to previous version...",
|
|
file=sys.stderr
|
|
)
|
|
|
|
# Restore our previous copy (if available).
|
|
if os.path.isfile((args.server_path + '.bak')):
|
|
os.rename((args.server_path + '.bak'), args.server_path)
|
|
|
|
if args.post_hook and not _run_hook(args.post_hook):
|
|
sys.exit("Post-hook command failed.")
|
|
|
|
sys.exit(status_code)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|