jenkins-utils/upload-pkg.py

#!/usr/bin/python3
# -*- coding=utf-8
from pyalpm import Handle
from sys import argv, exit, stdout
from os import listdir
from os.path import exists, basename, join
from logging import warning, info, basicConfig, INFO
from requests import get, post, put, delete, RequestException, Response
from argparse import ArgumentParser

default_server = "https://repo-updater.classfun.cn"


def FormatSize(size: int) -> str:
	units = ['Bytes', 'KB', 'MB', 'GB', 'TB', 'GB']
	unit = units[0]
	for unit in units:
		if size < 1024:
			break
		size /= 1024
	return "{:.2f} {}".format(size, unit)


def GetReasonText(res: Response, default: str = "Unknown") -> str:
	if "text/plain" not in res.headers["Content-Type"]:
		return default
	length = res.headers["Content-Length"]
	if length is None or int(length) >= 128:
		return default
	return res.text.strip()


def UploadFile(url: str, file: str, code: int = 201):
	with open(file, "rb") as f:
		res = put(url, f.read())
		text = GetReasonText(res)
		if res.status_code != code:
			raise RequestException(
				"upload %s status not %d: %d (%s)" %
				(file, code, res.status_code, text)
			)


def UploadPackage(
	pkg: str,
	sign: str = None,
	arch: str = None,
	server: str = default_server,
	handle: Handle = None
):
	if sign is None:
		sign = pkg + ".sig"
	if handle is None:
		handle = Handle("/", "/var/lib/pacman")
	if not exists(pkg):
		raise FileNotFoundError("Target package file %s not found" % pkg)
	if not exists(sign):
		raise FileNotFoundError("Target signature file %s not found" % pkg)
	if pkg + ".sig" != sign:
		warning("signature filename mismatch with package file name")
	alpm = handle.load_pkg(pkg)
	if not alpm:
		raise IOError("Open package %s failed" % pkg)
	info("package name: %s" % alpm.name)
	info("package version: %s" % alpm.version)
	info("package architecture: %s" % alpm.arch)
	info("package packager: %s" % alpm.packager)
	info("package size: %s" % FormatSize(alpm.size))
	info("package installed size: %s" % FormatSize(alpm.isize))
	info("package url: %s" % alpm.url)
	name = "%s-%s-%s" % (alpm.name, alpm.version, alpm.arch)
	res = get("%s/api/info" % server)
	if arch is None:
		arch = alpm.arch
	if arch == "any":
		raise ValueError("Unable to detect target architecture")
	if res.status_code != 200:
		raise RequestException("status not 200: %d" % res.status_code)
	base = res.json()
	pkg_ext = next((i for i in base["pkg_exts"] if pkg.endswith(i)), None)
	sign_ext = next((i for i in base["sign_exts"] if sign.endswith(i)), None)
	if pkg_ext is None:
		raise ValueError("Unknown package type")
	if sign_ext is None:
		raise ValueError("Unknown signature type")
	if arch not in base["arch"]:
		raise ValueError("Target architecture not found")
	pkg_name = name + pkg_ext
	sign_name = name + sign_ext
	if pkg_name != basename(pkg):
		warning("package filename mismatch with metadata")
	info("upload as %s" % pkg_name)
	try:
		UploadFile("%s/%s" % (server, sign_name), sign)
		UploadFile("%s/%s" % (server, pkg_name), pkg)
		res = post(
			"%s/api/update" % server,
			json={"arch": arch, "target": pkg_name},
			headers={"Content-Type": "application/json"}
		)
		text = GetReasonText(res)
		if res.status_code != 200:
			raise RequestException(
				"update status not 201: %d (%s)" %
				(res.status_code, text)
			)
		info("upload done")
	finally:
		delete("%s/%s" % (server, pkg_name))
		delete("%s/%s" % (server, sign_name))


def main(args: list) -> int:
	prs = ArgumentParser("Renegade Project Arch Linux Repo Uploader")
	prs.add_argument("-a", "--arch", help="Target repo architecture", required=False)
	prs.add_argument("-d", "--dir",  help="Package folder",           required=False)
	prs.add_argument("-p", "--pkg",  help="Package tarball file",     required=False)
	prs.add_argument("-s", "--sign", help="Package signature file",   required=False)
	prs.add_argument("-u", "--url",  help="Updater Server URL",       required=False, default=default_server)
	ps = prs.parse_args(args[1:])
	basicConfig(level=INFO, stream=stdout)
	cnt = 0
	if ps.pkg:
		UploadPackage(ps.pkg, ps.sign, ps.arch, ps.url)
		cnt += 1
	elif ps.dir:
		exts = [".pkg.tar.gz", ".pkg.tar.xz", ".pkg.tar.zst"]
		for f in listdir(ps.dir):
			full = join(ps.dir, f)
			if any(f.endswith(ext) for ext in exts):
				UploadPackage(full, None, ps.arch, ps.url)
				cnt += 1
	if cnt <= 0:
		raise Exception("no any package found")
	return 0


if __name__ == '__main__':
	exit(main(argv))