jenkins-utils/upload-pkg.py

146 lines
4.5 KiB
Python
Raw Normal View History

2023-12-14 14:51:59 +08:00
#!/usr/bin/python3
# -*- coding=utf-8
2023-12-14 12:24:23 +08:00
from pyalpm import Handle
from sys import argv, exit, stdout
2023-12-14 14:51:59 +08:00
from os import listdir
from os.path import exists, basename, join
2023-12-14 12:24:23 +08:00
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)
2023-12-21 15:11:21 +08:00
if res.status_code != code :
if res.status_code == 409:
info("target already exists; not overwritting")
2023-12-21 15:13:19 +08:00
else:
2023-12-21 15:11:21 +08:00
raise RequestException(
"upload %s status not %d: %d (%s)" %
(file, code, res.status_code, text)
)
2023-12-14 12:24:23 +08:00
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:
2023-12-21 15:19:50 +08:00
if res.status_code == 409:
info("target already exists; not overwritting")
else:
raise RequestException(
"update status not 201: %d (%s)" %
(res.status_code, text)
)
2023-12-14 12:24:23 +08:00
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)
2023-12-14 14:51:59 +08:00
prs.add_argument("-d", "--dir", help="Package folder", required=False)
prs.add_argument("-p", "--pkg", help="Package tarball file", required=False)
2023-12-14 12:24:23 +08:00
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)
2023-12-14 14:51:59 +08:00
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")
2023-12-14 12:24:23 +08:00
return 0
if __name__ == '__main__':
exit(main(argv))