Script for triggering or uploading Luanti ContentDB release.

luanti-cdb-release.py

#!/usr/bin/env python3

'''
MIT License

Copyright © 2026 Jordan Irwin (AntumDeluge)

Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
of the Software, and to permit persons to whom the Software is furnished to do
so, subject to the following conditions:

  The above copyright notice and this permission notice shall be included in
  all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
'''

'''
TODO:
- use built-in Python libraries for posting/uploading instead of external `curl`
'''


import argparse
import errno
import os
import json
import re
import subprocess
import sys


## Prints usage & help information.
def printUsageHelp():
	h_text = argp.format_help()

	overrides = [
		(r" \(default: None\)$", "")
	]
	for o in overrides:
		h_text = re.sub(o[0], o[1], h_text, flags=re.M)

	# titleize section headers
	h_text = re.sub(r"^usage: ", "Usage:\n  ", h_text, 1, re.M)
	h_text = re.sub(r"^options:", "Options:", h_text, 1, re.M)

	# move description to top
	h_text = re.sub(r"\n{}\n".format(argp.description), "", h_text, 1, re.M)
	h_text = argp.description + "\n\n" + h_text

	# change parameter strings from {val1,val2,...} to <val1|val2|...>
	h_text = re.sub(r"\{(.*?)\}", lambda m: f"<{m.group(1).replace(",", "|")}>", h_text, flags=re.M)

	lines = h_text.split("\n")
	for idx in range(len(lines)-1):
		l = lines[idx]
		# reformat parameter lines with arguments
		if l.startswith("  -") and re.search(r"<.*>", l):
			m = re.search(r"(?<!-)-(?!-).", l)
			s_short = m.group(0).lstrip("-") if m else None

			m = re.search(r"--(.*?)[ ,\n]", l)
			s_long = m.group(1).lstrip("-") if m else None

			m = re.search(r"<.*?>", l)
			params = m.group(0) if m else None

			# part of the description may be on the same line
			rem = re.sub(r".*>", "", l)

			if params:
				s_short = f"-{s_short}" if s_short else None
				s_long = f"--{s_long}" if s_long else None
				temp = ""
				for s in (s_short, s_long):
					if not s:
						continue
					if len(temp) > 0:
						temp = temp + ", "
					temp = temp + s
				temp = f"  {temp} {params}"
				if rem:
					l_diff = len(l) - (len(temp) + len(rem))
					if l_diff > 0:
						# pad to keep indentation of original line
						rem = (" " * l_diff) + rem

					temp = temp + rem
				lines[idx] = temp

	# print the formatted output
	print("\n".join(lines))


## Prints warning message to stdout.
#
#  @param msg
#    Text to be printed.
#  @param usage
#    If `True`, prints usage info after message.
def printWarn(msg, usage=False):
	if not msg.endswith("\n"):
		msg = f"{msg}\n"
	sys.stdout.write(f"\n== WARNING ==\n{msg}")
	if usage:
		print()
		printUsageHelp()


## Prints error message to stderr.
#
#  @param msg
#    Text to be printed.
#  @param usage
#    If `True`, prints usage info after message.
def printError(msg, usage=False):
	if not msg.endswith("\n"):
		msg = f"{msg}\n"
	sys.stderr.write(f"\n== ERROR ==\n{msg}")
	if usage:
		print()
		printUsageHelp()


## Prints message to stderr & exits.
#
#  @param e
#    Error code.
#  @param msg
#    Text to be printed.
#  @param usage
#    If `True`, prints usage info after message.
def exitWithError(e, msg, usage=False):
	printError(msg, usage)
	sys.exit(e)


## Initializes globals.
def init():
	global root, dir_conf

	if "root" not in globals():
		root = os.getcwd()
	if "dir_conf" not in globals():
		dir_conf = os.path.join(root, ".conf")


## Executes a system command as a subprocess.
#
#  @param args
#    Command & arguments to execute.
#  @return
#    Value of command output (from stdout).
def execute(*args):
	proc = subprocess.run(args, capture_output=True)
	if proc.returncode != 0:
		exitWithError(proc.returncode, proc.stderr.decode("UTF-8"))

	return proc.stdout.decode("UTF-8").strip("\n")


## Retrieves ContentDB API token.
#
#  Token is extracted from binary PGP encoded file `.contentdb/token.gpg.bin` using `gpg`. If the
#  token file doesn't exist, raw token text will be request from user input.
#
#  The token file can be created by doing following from project root:
#  - execute `gpg -e -r <email> -o .contentdb/token.gpg.bin`
#  - type in the raw token text
#  - press Ctrl+D (may need to press it twice) or press Enter then Ctrl+D
#
#  @return
#    Raw token text value.
def getContentDBToken():
	file_token = os.path.join(dir_conf, "cdb_token.gpg.bin")
	slug = os.path.join(*file_token.split(os.sep)[-2:])
	print(f"\nChecking for ContentDB API secret token file: {slug} ...")

	if not os.path.isfile(file_token):
		# user must input manually
		return input("\nInput ContentDB API token: ")

	print("Decoding ContentDB API secret token from file ...")
	return execute("gpg", "-d", file_token)


## Parses the most recent Git tag name.
#
#  @return
#    Tag name.
def parseGitTag():
	tag = execute("git", "describe", "--tags", "--abbrev=0")
	if tag == "":
		exitWithError(errno.EINVAL, "Tag for release not found")
	return tag


## Represents a repostory's configuration file located at `.conf/cdb_release.conf`.
class CDBConfig:
	# cached config keys & values
	data = {}
	# determines if project files are located in Git repository
	git = False


	## Initializes configuration values.
	@staticmethod
	def init():
		CDBConfig.path = os.path.join(dir_conf, "cdb_release.conf")
		CDBConfig.git = os.path.isdir(os.path.join(root, ".git"))
		CDBConfig._validate()
		CDBConfig._parseReleaseNotes()


	## Parses & validates configuration file.
	@staticmethod
	def _validate():
		# check that the configuration file is value
		if not os.path.exists(CDBConfig.path):
			exitWithError(errno.ENOENT, f"Cannot parse configuration, file not found: {CDBConfig.path}")
		if os.path.isdir(CDBConfig.path):
			exitWithError(errno.EISIDR, f"Cannot parse configuration, directory found: {CDBConfig.path}")

		# read config from disk
		print("\nParsing ContentDB config ...")
		fin = open(CDBConfig.path)
		lines = fin.readlines()
		fin.close()

		lidx = -1
		for l in lines:
			lidx = lidx + 1
			# ignore commented-out lines
			if l.startswith("#") or l.strip() == "":
				continue
			elif "=" not in l:
				print(f"\nWARNING: malformed config line ({lidx}): {l}")
				continue
			tmp = l.split("=", 1)
			CDBConfig._set(tmp[0], tmp[1])

		# use Git tag if release name not set
		if not CDBConfig.has("release") and CDBConfig.git:
			CDBConfig._set("release", parseGitTag())

		# check for mandatory keys
		for key in ("username", "package", "release"):
			if not CDBConfig.has(key):
				exitWithError(errno.ENOENT,
						f"Missing mandatory configuration key '{key}' or invalid value: {CDBConfig.path}")


	## Parses release notes from text file `.contentdb/notes.txt` & adds to configuration.
	@staticmethod
	def _parseReleaseNotes():
		notes = ""
		if args.notes:
			notes = args.notes
		else:
			notes_path = os.path.join(dir_conf, "cdb_release_notes.txt")
			if os.path.isfile(notes_path):
				fin = open(notes_path, "r", encoding="utf-8")
				notes = fin.read().strip()
				fin.close()

		# convert literal "\n" to newline character
		notes = notes.replace("\\n", "\n")

		CDBConfig._set("notes", notes)


	## Checks if a configuration key has been set.
	#
	#  @param key
	#    Key to be parsed.
	#  @return
	#    `True` if &lt;key&gt; found in config cache.
	@staticmethod
	def has(key):
		return CDBConfig.get(key, False) != None


	## Caches a configuration key=value pair (should only be used internally).
	#
	#  @param key
	#    Key index.
	#  @param value
	#    Stored value.
	@staticmethod
	def _set(key, value):
		CDBConfig.data[key.strip()] = value.strip()


	## Retrieves value from configuration.
	#
	#  @param key
	#    Key to be parsed.
	#  @param require
	#    If `True`, exit with error if key not found. (default: `True`)
	#  @return
	#    Parsed value from config or `None`.
	@staticmethod
	def get(key, require=True):
		if key not in CDBConfig.data:
			if require:
				exitWithError(errno.ENOENT, f"Configuration key not found: {key}")
			else:
				return None

		value = CDBConfig.data[key]
		if value == "":
			if require:
				exitWithError(errno.ENOENT, f"Empty configuration key: {key}")
			else:
				return None

		return value


	## Retrieves release URL path.
	#
	#  @return
	#    Release URL, username, & package name.
	@staticmethod
	def getReleaseURL():
		username, package = CDBConfig.get("username"), CDBConfig.get("package")
		return f"https://content.luanti.org/api/packages/{username}/{package}/releases", \
				username, package

	## Parses info to be used in release.
	#
	#  @return
	#    Username, package name, release name, & URL to post release.
	@staticmethod
	def getReleaseInfo():
		url, username, package = CDBConfig.getReleaseURL()
		url = url + "/new/"
		return username, package, CDBConfig.get("release"), url



## Retrieves list of current releases from ContentDB.
#
#  @return
#    Table of current releases info.
def getCurrentReleases():
	url, username, package = CDBConfig.getReleaseURL()
	print(f"\nContacting ContentDB for releases information of package {username}/{package} ...")
	try:
		return json.loads(execute("curl", "-X", "GET", "-L", url))
	except json.decoder.JSONDecodeError as e:
		exitWithError(1, "failed to retrieve releases information")


## Retrieves a list of current releases names from ContentDB.
#
#  @return
#    List of current releases names.
def getCurrentReleasesNames():
	releases = []
	for r in getCurrentReleases():
		releases.append(r["name"])
	return releases


## Checks for release name usability.
#
#  If check fails, process exits with error.
#
#  @param name
#    Name being used for new release.
def checkReleaseName(name):
	if name in getCurrentReleasesNames():
		exitWithError(errno.EEXIST, f"release name already exists: {name}")


## Creates a distribution package.
#
#  @param ref Reference from which to package files.
def createRelease(ref):
	print("\nCreating distribution package ...")

	dir_target = os.path.join(root, "release")
	if not os.path.isdir(dir_target):
		if os.path.exists(dir_target):
			exitWithError(errno.EEXIST, "Cannot create directory, file exists: {}".format(dir_target))
		os.makedirs(dir_target)

	file_target = os.path.join(dir_target, "{}.zip".format(ref))
	if os.path.exists(file_target):
		exitWithError(errno.EEXIST, "Cannot create archive, file or directory exists: {}".format(file_target))

	execute("git", "archive", "--format=zip", "--output={}".format(file_target), ref)
	if not os.path.isfile(file_target):
		exitWithError(errno.ENOENT, "An unknown error occurred when creating distribution archive")
	print("Distribution archive created: {}".format(file_target))

	return file_target


## Uploads & creates ContentDB release from local distribution package.
def doUploadRelease():
	username, package, tag, url = CDBConfig.getReleaseInfo()
	name = args.name or tag
	checkReleaseName(name)
	title = args.title or name
	commit = execute("git", "rev-parse", tag)
	notes = CDBConfig.get("notes", False) or ""

	file_release = createRelease(tag)

	params = [
		"curl", "-X", "POST", url,
		"-H", "Authorization: Bearer {}".format(getContentDBToken()),
		"-F", f'name="{name}"',
		"-F", f'title="{title}"',
		"-F", f'commit="{commit}"',
		"-F", f'file="{file_release}"'
	]
	if notes:
		params.append("-F")
		params.append(f'release_notes="{notes}"')

	print("\n=== RELEASE INFO ===" + \
			f"\npackage: {username}/{package}" + \
			f"\nref:     {commit}" + \
			f"\nname:    {name}" + \
			f"\ntitle:   {title}")

	print("\n<<< RELEASE NOTES START >>>")
	print(notes)
	print("<<< RELEASE NOTES END >>>")

	confirmed = input("\nUpload ContentDB release? (y/n) ").lower() in ("y", "yes")
	if confirmed:
		print("\nUploading release archive ...")
		execute(*params)
	else:
		print("\nCancelled release")


## Tells ContentDB to scan remote Git repo for new release.
def doGitRelease():
	username, package, tag, url = CDBConfig.getReleaseInfo()
	name = args.name or tag
	checkReleaseName(name)
	title = args.title or name
	commit = execute("git", "rev-parse", tag)
	notes = CDBConfig.get("notes", False) or ""

	params = [
		"curl", "-X", "POST", url,
		"-H", "Authorization: Bearer {}".format(getContentDBToken()),
		"-H", "Content-Type: application/json",
		"-d", f'{{"method": "git", "name": "{name}", "title": "{title}", "ref": "{commit}", "release_notes": "{notes}"}}'
	]

	print("\n=== RELEASE INFO ===" + \
			f"\npackage: {username}/{package}" + \
			f"\nref:     {commit}" + \
			f"\nname:    {name}" + \
			f"\ntitle:   {title}")

	print("\n<<< RELEASE NOTES START >>>")
	print(notes)
	print("<<< RELEASE NOTES END >>>")

	confirmed = input("\nPost ContentDB release? (y/n) ").lower() in ("y", "yes")
	if confirmed:
		print("\nPosting release info ...")
		execute(*params)
	else:
		print("\nCancelled release")


## Parses command line parameters.
def parseCommandLine():
	global args, argp

	argp = argparse.ArgumentParser(
			description="Script for creating Luanti ContentDB releases.",
			formatter_class=argparse.ArgumentDefaultsHelpFormatter
	)
	argp.add_argument("-m", "--mode", action="store", default="git", \
			choices=("git", "upload"), \
			help="Mode of release. By default, releases will be created from remote Git repository. " \
				+ "But if this value is set to \"upload\", a release package distribution file will be " \
				+ "created & uploaded instead.")
	argp.add_argument("-n", "--name", action="store", metavar="<name>", help="Release name. (default: use Git tag)")
	argp.add_argument("-t", "--title", action="store", metavar="<title>", help="Release title. (default: use name)")
	argp.add_argument("--notes", action="store", metavar="<notes>", \
			help="Release notes. (default: parse .conf/cdb_release_notes.txt)")

	# override help output method
	print_help_orig = argp.print_help
	argp.print_help = printUsageHelp

	args = argp.parse_args(sys.argv[1:])


if __name__ != "__main__":
	exitWithError(errno.ENOEXEC, f"cannot import {os.path.basename(__file__)} as a library")
else:
	parseCommandLine()
	if "args" not in globals() or "argp" not in globals():
		exitWithError(1, "failed to parse command line")

	# initialize path globals
	init()
	# initialize & config
	CDBConfig.init()

	if args.mode == "git":
		doGitRelease()
	elif args.mode == "upload":
		doUploadRelease()
	else:
		exitWithError(errno.EINVAL, f"unknown mode \"{args.mode}\"", True)