--- /dev/null
+#!/usr/bin/env python3
+
+import os
+import stat
+import sys
+import argparse
+import subprocess
+import re
+from packaging import version
+import atexit
+import glob
+import toml
+import datetime
+import time
+import shutil
+import getpass
+import tempfile
+
+SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
+SUDO_PROC = False
+AUR_GIT_REPO_PATH = "https://aur.archlinux.org"
+AUR_GIT_REPO_PATH_TEMPLATE = AUR_GIT_REPO_PATH + "/{}.git"
+global GLOBAL_LOG_FILE
+GLOBAL_LOG_FILE = "log.txt"
+
+
+def log_print(string):
+ print(string)
+ with open(GLOBAL_LOG_FILE, "a", encoding="utf-8") as lf:
+ print(string, file=lf)
+
+
+def ensure_pkg_dir_exists(pkg, pkg_state):
+ log_print('Checking that dir for "{}" exists...'.format(pkg))
+ pkg_dir = os.path.join(SCRIPT_DIR, pkg)
+ if os.path.isdir(pkg_dir):
+ log_print('Dir for "{}" exists.'.format(pkg))
+ return True
+ elif os.path.exists(pkg_dir):
+ log_print('"{}" exists but is not a dir'.format(pkg_dir))
+ return False
+ elif "repo_path" not in pkg_state[pkg]:
+ pkg_state[pkg]["repo_path"] = AUR_GIT_REPO_PATH_TEMPLATE.format(pkg)
+ try:
+ subprocess.run(
+ ["git", "clone", pkg_state[pkg]["repo_path"], pkg],
+ check=True,
+ cwd=SCRIPT_DIR,
+ )
+ except subprocess.CalledProcessError:
+ print(
+ 'ERROR: Failed to git clone "{}" (tried repo path "{}")'.format(
+ pkg_dir, pkg_state[pkg]["repo_path"]
+ )
+ )
+ return False
+ log_print('Created dir for "{}".'.format(pkg))
+ return True
+ elif pkg_state[pkg]["repo_path"] == "NO_REPO":
+ log_print('"{}" does not exist, but NO_REPO specified for repo_path')
+ return False
+
+
+def update_pkg_dir(pkg, state):
+ log_print('Making sure pkg dir for "{}" is up to date...'.format(pkg))
+
+ # fetch all
+ try:
+ subprocess.run(
+ ["git", "fetch", "-p", "--all"],
+ check=True,
+ cwd=os.path.join(SCRIPT_DIR, pkg),
+ )
+ except subprocess.CalledProcessError:
+ log_print(
+ 'ERROR: Failed to update pkg dir of "{}" (fetching).'.format(pkg)
+ )
+ return False, False
+
+ # get remotes
+ remotes = []
+ try:
+ result = subprocess.run(
+ ["git", "remote"],
+ check=True,
+ cwd=os.path.join(SCRIPT_DIR, pkg),
+ capture_output=True,
+ encoding="UTF-8",
+ )
+ remotes = result.stdout.split(sep="\n")
+ except subprocess.CalledProcessError:
+ log_print(
+ 'ERROR: Failed to update pkg dir of "{}" (getting remotes).'.format(
+ pkg
+ )
+ )
+ return False, False
+ remotes = list(filter(lambda s: len(s) > 0, remotes))
+ if len(remotes) == 0:
+ log_print(
+ 'ERROR: Failed to update pkg dir of "{}" (getting remotes).'.format(
+ pkg
+ )
+ )
+ return False, False
+
+ # get remote that current branch is tracking
+ selected_remote = ""
+ try:
+ result = subprocess.run(
+ ["git", "status", "-sb", "--porcelain"],
+ check=True,
+ cwd=os.path.join(SCRIPT_DIR, pkg),
+ capture_output=True,
+ encoding="UTF-8",
+ )
+ for remote in remotes:
+ if (
+ len(remote.strip()) > 0
+ and result.stdout.find(remote.strip()) != -1
+ ):
+ selected_remote = remote.strip()
+ break
+ except subprocess.CalledProcessError:
+ print(
+ 'ERROR: Failed to update pkg dir of "{}" (getting branch\'s remote).'.format(
+ pkg
+ )
+ )
+ return False, False
+ if len(selected_remote) == 0:
+ print(
+ 'ERROR: Failed to update pkg dir of "{}" (getting branch\'s remote).'.format(
+ pkg
+ )
+ )
+ return False, False
+
+ # get hash of current branch
+ current_branch_hash = ""
+ try:
+ result = subprocess.run(
+ ["git", "log", "-1", "--format=format:%H"],
+ check=True,
+ cwd=os.path.join(SCRIPT_DIR, pkg),
+ capture_output=True,
+ encoding="UTF-8",
+ )
+ current_branch_hash = result.stdout.strip()
+ except subprocess.CalledProcessError:
+ print(
+ 'ERROR: Failed to update pkg dir of "{}" (getting current branch\'s hash).'.format(
+ pkg
+ )
+ )
+ return False, False
+ if len(current_branch_hash.strip()) == 0:
+ print(
+ 'ERROR: Failed to update pkg dir of "{}" (getting current branch\'s hash).'.format(
+ pkg
+ )
+ )
+ return False, False
+
+ # get hash of remote branch
+ remote_branch_hash = ""
+ try:
+ result = subprocess.run(
+ ["git", "log", "-1", "--format=format:%H", selected_remote],
+ check=True,
+ cwd=os.path.join(SCRIPT_DIR, pkg),
+ capture_output=True,
+ encoding="UTF-8",
+ )
+ remote_branch_hash = result.stdout.strip()
+ except subprocess.CalledProcessError:
+ print(
+ 'ERROR: Failed to update pkg dir of "{}" (getting remote branch\'s hash).'.format(
+ pkg
+ )
+ )
+ return False, False
+ if len(remote_branch_hash.strip()) == 0:
+ print(
+ 'ERROR: Failed to update pkg dir of "{}" (getting remote branch\'s hash).'.format(
+ pkg
+ )
+ )
+ return False, False
+
+ # update current branch if not same commit
+ if current_branch_hash != remote_branch_hash:
+ try:
+ subprocess.run(
+ ["git", "pull"], check=True, cwd=os.path.join(SCRIPT_DIR, pkg)
+ )
+ except subprocess.CalledProcessError:
+ try:
+ subprocess.run(
+ ["git", "checkout", "--", "*"],
+ check=True,
+ cwd=os.path.join(SCRIPT_DIR, pkg),
+ )
+ subprocess.run(
+ ["git", "pull"],
+ check=True,
+ cwd=os.path.join(SCRIPT_DIR, pkg),
+ )
+ except subprocess.CalledProcessError:
+ log_print(
+ 'ERROR: Failed to update pkg dir of "{}".'.format(pkg)
+ )
+ return False, False
+ elif state[pkg]["skip_branch_up_to_date"]:
+ log_print(f'"{pkg}" is up to date')
+ return True, True
+ log_print('Updated pkg dir for "{}"'.format(pkg))
+ return True, False
+
+
+def check_pkg_build(pkg, editor):
+ """Returns "ok", "not_ok", "abort", or "force_build"."""
+ log_print('Checking PKGBUILD for "{}"...'.format(pkg))
+ try:
+ subprocess.run(
+ [editor, os.path.join(pkg, "PKGBUILD")], check=True, cwd=SCRIPT_DIR
+ )
+ except subprocess.CalledProcessError:
+ log_print('ERROR: Failed checking PKGBUILD for "{}"'.format(pkg))
+ return "abort"
+ while True:
+ log_print(
+ "PKGBUILD okay? [Y/n/c(heck again)/a(bort)/f(orce build)/b(ack)]"
+ )
+ user_input = sys.stdin.buffer.readline().decode().strip().lower()
+ if user_input == "y" or len(user_input) == 0:
+ log_print("User decided PKGBUILD is ok")
+ return "ok"
+ elif user_input == "n":
+ log_print("User decided PKGBUILD is not ok")
+ return "not_ok"
+ elif user_input == "c":
+ log_print("User will check PKGBUILD again")
+ return check_pkg_build(pkg, editor)
+ elif user_input == "a":
+ return "abort"
+ elif user_input == "f":
+ return "force_build"
+ elif user_input == "b":
+ return "back"
+ else:
+ log_print("ERROR: User gave invalid input...")
+ continue
+
+
+def check_pkg_version(pkgdir, pkg_state, repo, force_check_srcinfo):
+ """Returns "fail", "install", or "done"."""
+ status, current_epoch, current_version = get_pkg_current_version(pkgdir, pkg_state, repo)
+ if status != "fetched":
+ return status
+ elif current_version is None:
+ print(
+ 'ERROR: Failed to get version from package "{}".'.format(
+ pkg_state[pkgdir]["pkg_name"]
+ )
+ )
+ return "fail"
+ print(
+ 'Got version "{}:{}" for installed pkg "{}"'.format(
+ current_epoch if current_epoch is not None else "0",
+ current_version,
+ pkg_state[pkgdir]["pkg_name"],
+ )
+ )
+
+ return get_srcinfo_check_result(current_epoch, current_version, pkgdir, force_check_srcinfo)
+
+
+def get_srcinfo_version(pkgdir):
+ """Returns (success_bool, pkgepoch, pkgver, pkgrel)"""
+ if not os.path.exists(os.path.join(SCRIPT_DIR, pkgdir, ".SRCINFO")):
+ log_print(f'ERROR: .SRCINFO does not exist for pkg "{pkgdir}"')
+ return False, None, None, None
+ pkgver_reprog = re.compile(
+ "^\\s*pkgver\\s*=\\s*([a-zA-Z0-9._+-]+)\\s*$"
+ )
+ pkgrel_reprog = re.compile("^\\s*pkgrel\\s*=\\s*([0-9.]+)\\s*$")
+ pkgepoch_reprog = re.compile("^\\s*epoch\\s*=\\s*([0-9]+)\\s*$")
+ pkgver = ""
+ pkgrel = ""
+ pkgepoch = ""
+ with open(
+ os.path.join(SCRIPT_DIR, pkgdir, ".SRCINFO"), encoding="UTF-8"
+ ) as fo:
+ line = fo.readline()
+ while len(line) > 0:
+ pkgver_result = pkgver_reprog.match(line)
+ pkgrel_result = pkgrel_reprog.match(line)
+ pkgepoch_result = pkgepoch_reprog.match(line)
+ if pkgver_result:
+ pkgver = pkgver_result.group(1)
+ elif pkgrel_result:
+ pkgrel = pkgrel_result.group(1)
+ elif pkgepoch_result:
+ pkgepoch = pkgepoch_result.group(1)
+ line = fo.readline()
+ return True, pkgepoch, pkgver, pkgrel
+
+
+def get_pkgbuild_version(pkgdir, force_check_srcinfo):
+ """Returns (success, epoch, version, release)"""
+ log_print('Getting PKGBUILD version of "{}"...'.format(pkgdir))
+ while True and not force_check_srcinfo:
+ log_print("Use .SRCINFO or directly parse PKGBUILD?")
+ user_input = input("1 for .SRCINFO, 2 for PKGBUILD > ")
+ if user_input == "1" or user_input == "2":
+ break
+ # TODO support split packages
+ if force_check_srcinfo or user_input == "1":
+ srcinfo_fetch_success, pkgepoch, pkgver, pkgrel = get_srcinfo_version(pkgdir)
+ if not srcinfo_fetch_success:
+ log_print('ERROR: Failed to get pkg info from .SRCINFO')
+ return False, None, None, None
+ elif user_input == "2":
+ try:
+ log_print(
+ 'Running "makepkg --nobuild" to ensure pkgver in PKGBUILD is updated...'
+ )
+ subprocess.run(
+ ["makepkg", "-c", "--nobuild", "-s", "-r"],
+ check=True,
+ cwd=os.path.join(SCRIPT_DIR, pkgdir),
+ )
+ except subprocess.CalledProcessError:
+ log_print(
+ 'ERROR: Failed to run "makepkg --nobuild" in "{}".'.format(
+ pkgdir
+ )
+ )
+ if os.path.exists(os.path.join(SCRIPT_DIR, pkgdir, "src")):
+ shutil.rmtree(os.path.join(SCRIPT_DIR, pkgdir, "src"))
+ return False, None, None, None
+
+ if os.path.exists(os.path.join(SCRIPT_DIR, pkgdir, "src")):
+ shutil.rmtree(os.path.join(SCRIPT_DIR, pkgdir, "src"))
+ pkgepoch = ""
+ pkgver = ""
+ pkgrel = ""
+
+ # TODO maybe sandbox sourcing the PKGBUILD
+ pkgbuild_output = subprocess.run(
+ [
+ "bash",
+ "-c",
+ f"source {os.path.join(SCRIPT_DIR, pkgdir, 'PKGBUILD')}; echo \"pkgver=$pkgver\"; echo \"pkgrel=$pkgrel\"; echo \"epoch=$epoch\"",
+ ],
+ capture_output=True,
+ text=True,
+ )
+ output_ver_re = re.compile(
+ "^pkgver=([a-zA-Z0-9._+-]+)\\s*$", flags=re.M
+ )
+ output_rel_re = re.compile("^pkgrel=([0-9.]+)\\s*$", flags=re.M)
+ output_epoch_re = re.compile("^epoch=([0-9]+)\\s*$", flags=re.M)
+
+ match = output_ver_re.search(pkgbuild_output.stdout)
+ if match:
+ pkgver = match.group(1)
+ match = output_rel_re.search(pkgbuild_output.stdout)
+ if match:
+ pkgrel = match.group(1)
+ match = output_epoch_re.search(pkgbuild_output.stdout)
+ if match:
+ pkgepoch = match.group(1)
+ else:
+ log_print("ERROR: Unreachable code")
+ return False, None, None, None
+
+ if len(pkgepoch) == 0:
+ pkgepoch = None
+ if len(pkgver) == 0:
+ pkgver = None
+ if len(pkgrel) == 0:
+ pkgrel = None
+
+ if pkgver is not None and pkgrel is not None:
+ return True, pkgepoch, pkgver, pkgrel
+ else:
+ log_print(
+ 'ERROR: Failed to get PKGBUILD version of "{}".'.format(pkgdir)
+ )
+ return False, None, None, None
+
+
+def get_srcinfo_check_result(current_epoch, current_version, pkgdir, force_check_srcinfo):
+ ver_success, pkgepoch, pkgver, pkgrel = get_pkgbuild_version(pkgdir, force_check_srcinfo)
+ if ver_success:
+ if current_epoch is None and pkgepoch is not None:
+ print(
+ 'Current installed version of "{}" is out of date (missing epoch).'.format(
+ pkg_state[pkgdir]["pkg_name"]
+ )
+ )
+ return "install"
+ elif current_epoch is not None and pkgepoch is None:
+ print(
+ 'Current installed version of "{}" is up to date (has epoch).'.format(
+ pkg_state[pkgdir]["pkg_name"]
+ )
+ )
+ return "done"
+ elif current_epoch is not None and pkgepoch is not None and int(current_epoch) < int(pkgepoch):
+ print(
+ 'Current installed version of "{}" is out of date (older epoch).'.format(
+ pkg_state[pkgdir]["pkg_name"]
+ )
+ )
+ return "install"
+ elif pkgver is not None and pkgrel is not None and version.parse(current_version) < version.parse(pkgver + "-" + pkgrel):
+ print(
+ 'Current installed version of "{}" is out of date (older version).'.format(
+ pkg_state[pkgdir]["pkg_name"]
+ )
+ )
+ return "install"
+ else:
+ print(
+ 'Current installed version of "{}" is up to date.'.format(
+ pkg_state[pkgdir]["pkg_name"]
+ )
+ )
+ return "done"
+ else:
+ print(
+ 'ERROR: Failed to get pkg_version of "{}"'.format(
+ pkg_state[pkgdir]["pkg_name"]
+ )
+ )
+ return "fail"
+
+def get_pkg_current_version(pkgdir, pkg_state, repo):
+ """Returns (status, epoch, version)"""
+ print(
+ 'Checking version of installed pkg "{}"...'.format(
+ pkg_state[pkgdir]["pkg_name"]
+ )
+ )
+ current_epoch = None
+ current_version = None
+ try:
+ result = subprocess.run(
+ "tar -tf {} | grep '{}.*/$'".format(
+ repo, pkg_state[pkgdir]["pkg_name"]
+ ),
+ check=True,
+ capture_output=True,
+ encoding="UTF-8",
+ shell=True,
+ )
+ reprog = re.compile(
+ "^{}-(?P<epoch>[0-9]+:)?(?P<version>[^-/: ]*-[0-9]+)/$".format(
+ pkg_state[pkgdir]["pkg_name"]
+ ),
+ flags=re.MULTILINE,
+ )
+ reresult = reprog.search(result.stdout)
+ if reresult:
+ result_dict = reresult.groupdict()
+ if not result_dict["epoch"] is None:
+ current_epoch = result_dict["epoch"][:-1]
+ if not result_dict["version"] is None:
+ current_version = result_dict["version"]
+ else:
+ print(
+ "ERROR: Failed to get current version from repo for package {}".format(
+ pkg_state[pkgdir]["pkg_name"]
+ )
+ )
+ return "fail", None, None
+ except subprocess.CalledProcessError:
+ log_print("Package not found, assuming building first time.")
+ return "install", None, None
+ return "fetched", current_epoch, current_version
+
+
+def get_sudo_privileges():
+ global SUDO_PROC
+ if not SUDO_PROC:
+ log_print("sudo -v")
+ try:
+ subprocess.run(["sudo", "-v"], check=True)
+ except subprocess.CalledProcessError:
+ return False
+ SUDO_PROC = subprocess.Popen(
+ ["while true; do sudo -v; sleep 2m; done"], shell=True
+ )
+ atexit.register(cleanup_sudo, sudo_proc=SUDO_PROC)
+ return True
+ return True
+
+
+def cleanup_sudo(sudo_proc):
+ sudo_proc.terminate()
+
+
+def create_executable_script(dest_filename, script_contents):
+ tempf_name = "unknown"
+ with tempfile.NamedTemporaryFile(
+ mode="w", encoding="utf-8", delete=False
+ ) as tempf:
+ print(
+ """#!/usr/bin/env python3
+import os
+import stat
+import argparse
+
+def create_executable_script(dest_filename, script_contents):
+ with open(dest_filename, mode='w', encoding='utf-8') as f:
+ f.write(script_contents)
+ os.chmod(dest_filename, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR
+ | stat.S_IRGRP | stat.S_IXGRP
+ | stat.S_IROTH | stat.S_IXOTH)
+if __name__ == '__main__':
+ parser = argparse.ArgumentParser(description="Set new file with execute permissions")
+ parser.add_argument("--dest_filename")
+ parser.add_argument("--script_contents")
+ args = parser.parse_args()
+
+ create_executable_script(args.dest_filename, args.script_contents)
+""",
+ file=tempf,
+ )
+ tempf_name = tempf.name
+ try:
+ subprocess.run(
+ [
+ "sudo",
+ "/usr/bin/env",
+ "python3",
+ tempf_name,
+ "--dest_filename",
+ dest_filename,
+ "--script_contents",
+ script_contents,
+ ],
+ check=True,
+ )
+ except subprocess.CalledProcessError:
+ log_print(
+ f'ERROR: Failed to create executable script "{dest_filename}"'
+ )
+ return False
+ return True
+
+
+def setup_ccache(chroot):
+ # set up ccache stuff
+ try:
+ subprocess.run(
+ [
+ "sudo",
+ "sed",
+ "-i",
+ "/^BUILDENV=/s/!ccache/ccache/",
+ f"{chroot}/root/etc/makepkg.conf",
+ ],
+ check=True,
+ cwd=SCRIPT_DIR,
+ )
+ except subprocess.CalledProcessError:
+ log_print("ERROR: Failed to enable ccache in makepkg.conf")
+ sys.exit(1)
+
+
+def cleanup_ccache(chroot):
+ # cleanup ccache stuff
+ try:
+ subprocess.run(
+ [
+ "sudo",
+ "sed",
+ "-i",
+ "/^BUILDENV=/s/ ccache/ !ccache/",
+ f"{chroot}/root/etc/makepkg.conf",
+ ],
+ check=True,
+ cwd=SCRIPT_DIR,
+ )
+ except subprocess.CalledProcessError:
+ log_print("ERROR: Failed to disable ccache in makepkg.conf")
+ sys.exit(1)
+
+
+def setup_sccache(chroot):
+ sccache_script = """#!/usr/bin/env sh
+export PATH=${PATH/:\/usr\/local\/bin/}
+/usr/bin/env sccache $(basename "$0") "$@"
+"""
+ if (
+ not create_executable_script(
+ f"{chroot}/root/usr/local/bin/gcc", sccache_script
+ )
+ or not create_executable_script(
+ f"{chroot}/root/usr/local/bin/g++", sccache_script
+ )
+ or not create_executable_script(
+ f"{chroot}/root/usr/local/bin/clang", sccache_script
+ )
+ or not create_executable_script(
+ f"{chroot}/root/usr/local/bin/clang++", sccache_script
+ )
+ or not create_executable_script(
+ f"{chroot}/root/usr/local/bin/rustc", sccache_script
+ )
+ ):
+ log_print("ERROR: Failed to set up sccache wrapper scripts")
+ sys.exit(1)
+
+
+def cleanup_sccache(chroot):
+ # cleanup sccache stuff
+ try:
+ subprocess.run(
+ [
+ "sudo",
+ "rm",
+ "-f",
+ f"{chroot}/root/usr/local/bin/gcc",
+ f"{chroot}/root/usr/local/bin/g++",
+ f"{chroot}/root/usr/local/bin/clang",
+ f"{chroot}/root/usr/local/bin/clang++",
+ f"{chroot}/root/usr/local/bin/rustc",
+ ],
+ check=False,
+ cwd=SCRIPT_DIR,
+ )
+ except BaseException:
+ log_print("WARNING: Failed to cleanup sccache files")
+
+
+def update_pkg_list(
+ pkgs,
+ pkg_state,
+ chroot,
+ pkg_out_dir,
+ repo,
+ logs_dir,
+ no_update,
+ signing_gpg_dir,
+ signing_gpg_key_fp,
+ signing_gpg_pass,
+ no_store,
+):
+ if not get_sudo_privileges():
+ log_print("ERROR: Failed to get sudo privileges")
+ sys.exit(1)
+ if not no_update:
+ log_print("Updating the chroot...")
+ try:
+ subprocess.run(
+ ["arch-nspawn", "{}/root".format(chroot), "pacman", "-Syu"],
+ check=True,
+ )
+ except subprocess.CalledProcessError:
+ log_print("ERROR: Failed to update the chroot")
+ sys.exit(1)
+ for pkg in pkgs:
+ log_print(f'Building "{pkg}"...')
+ if "ccache_dir" in pkg_state[pkg]:
+ cleanup_sccache(chroot)
+ setup_ccache(chroot)
+ else:
+ cleanup_ccache(chroot)
+ if "sccache_dir" in pkg_state[pkg]:
+ setup_sccache(chroot)
+ else:
+ cleanup_sccache(chroot)
+
+ command_list = [
+ "makechrootpkg",
+ "-c",
+ "-r",
+ chroot,
+ ]
+ post_command_list = [
+ "--",
+ "--syncdeps",
+ "--noconfirm",
+ "--log",
+ "--holdver",
+ ]
+ for dep in pkg_state[pkg]["other_deps"]:
+ dep_fullpath = get_latest_pkg(dep, "/var/cache/pacman/pkg")
+ if not dep_fullpath:
+ log_print('ERROR: Failed to get dep "{}"'.format(dep))
+ sys.exit(1)
+ command_list.insert(1, "-I")
+ command_list.insert(2, dep_fullpath)
+ for aur_dep in pkg_state[pkg]["aur_deps"]:
+ aur_dep_fullpath = get_latest_pkg(aur_dep, pkg_out_dir)
+ if not aur_dep_fullpath:
+ log_print('ERROR: Failed to get aur_dep "{}"'.format(aur_dep))
+ sys.exit(1)
+ command_list.insert(1, "-I")
+ command_list.insert(2, aur_dep_fullpath)
+ if "ccache_dir" in pkg_state[pkg]:
+ command_list.insert(1, "-d")
+ command_list.insert(2, f'{pkg_state[pkg]["ccache_dir"]}:/ccache')
+ post_command_list.insert(1, "CCACHE_DIR=/ccache")
+ elif "sccache_dir" in pkg_state[pkg]:
+ command_list.insert(1, "-d")
+ command_list.insert(2, f'{pkg_state[pkg]["sccache_dir"]}:/sccache')
+ post_command_list.insert(1, "SCCACHE_DIR=/sccache")
+ post_command_list.insert(
+ 2, f'SCCACHE_CACHE_SIZE={pkg_state[pkg]["sccache_cache_size"]}'
+ )
+ post_command_list.insert(3, "RUSTC_WRAPPER=/usr/bin/sccache")
+ nowstring = datetime.datetime.now(datetime.timezone.utc).strftime(
+ "%Y-%m-%d_%H-%M-%S_%Z"
+ )
+ # log_print(f"Using command list: {command_list + post_command_list}") # DEBUG
+ with open(
+ os.path.join(logs_dir, "{}_stdout_{}".format(pkg, nowstring)),
+ "w",
+ ) as log_stdout, open(
+ os.path.join(logs_dir, "{}_stderr_{}".format(pkg, nowstring)),
+ "w",
+ ) as log_stderr:
+ try:
+ subprocess.run(
+ command_list + post_command_list,
+ check=True,
+ cwd=os.path.join(SCRIPT_DIR, pkg),
+ stdout=log_stdout,
+ stderr=log_stderr,
+ )
+ except subprocess.CalledProcessError:
+ log_print(
+ 'ERROR: Failed to build pkg "{}" in chroot'.format(pkg)
+ )
+ pkg_state[pkg]["build_status"] = "fail"
+ continue
+
+ if no_store:
+ pkg_state[pkg]["build_status"] = "success"
+ continue
+
+ pkg_list = glob.glob(os.path.join(SCRIPT_DIR, pkg, "*.pkg.tar*"))
+
+ log_print("Signing package...")
+ for gpkg in pkg_list:
+ try:
+ command_list = [
+ "gpg",
+ "--batch",
+ "--passphrase-fd",
+ "0",
+ "--pinentry-mode",
+ "loopback",
+ "--default-key",
+ signing_gpg_key_fp,
+ "--detach-sign",
+ gpkg,
+ ]
+ subprocess.run(
+ command_list,
+ check=True,
+ cwd=os.path.join(SCRIPT_DIR, pkg),
+ input=signing_gpg_pass,
+ text=True,
+ env={"GNUPGHOME": signing_gpg_dir},
+ )
+ except subprocess.CalledProcessError:
+ log_print(f'ERROR: Failed to sign pkg "{pkg}"')
+
+ log_print("Adding built pkgs to repo...")
+ try:
+ command_list = ["repo-add", repo]
+ for gpkg in pkg_list:
+ command_list.append(gpkg)
+ subprocess.run(command_list, check=True)
+ except subprocess.CalledProcessError:
+ log_print(
+ 'ERROR: Failed to add built pkg(s) "{}" to repo.'.format(pkg)
+ )
+ pkg_state[pkg]["build_status"] = "add_fail"
+ continue
+
+ log_print('Signing "custom.db"...')
+ try:
+ subprocess.run(
+ [
+ "/usr/bin/rm",
+ "-f",
+ str(os.path.join(pkg_out_dir, "custom.db.sig")),
+ ]
+ )
+ subprocess.run(
+ [
+ "/usr/bin/gpg",
+ "--batch",
+ "--passphrase-fd",
+ "0",
+ "--pinentry-mode",
+ "loopback",
+ "--default-key",
+ signing_gpg_key_fp,
+ "--detach-sign",
+ str(os.path.join(pkg_out_dir, "custom.db")),
+ ],
+ check=True,
+ input=signing_gpg_pass,
+ text=True,
+ env={"GNUPGHOME": signing_gpg_dir},
+ )
+ except subprocess.CalledProcessError:
+ log_print('WARNING: Failed to sign "custom.db"')
+
+ pkg_state[pkg]["build_status"] = "success"
+
+ log_print("Moving pkg to pkgs directory...")
+ for f in pkg_list:
+ log_print(f'Moving "{f}"...')
+ os.rename(f, os.path.join(pkg_out_dir, os.path.basename(f)))
+ sig_name = f + ".sig"
+ if os.path.exists(sig_name):
+ log_print(f'Moving "{sig_name}"...')
+ os.rename(
+ sig_name,
+ os.path.join(pkg_out_dir, os.path.basename(sig_name)),
+ )
+
+ for pkg in pkgs:
+ log_print(f'"{pkg}" status: {pkg_state[pkg]["build_status"]}')
+
+
+def get_latest_pkg(pkg, cache_dir):
+ globbed = glob.glob(os.path.join(cache_dir, pkg + "*"))
+ if len(globbed) > 0:
+ globbed.sort()
+ reprog = re.compile(
+ ".*"
+ + pkg
+ + "-[0-9a-zA-Z.+_:]+-[0-9a-zA-Z.+_]+-(any|x86_64).pkg.tar.(xz|gz|zst)$"
+ )
+ result = list(filter(lambda x: reprog.match(x), globbed))
+ if len(result) == 0:
+ return None
+ else:
+ return result[-1]
+ else:
+ return None
+
+
+def confirm_result(pkg, state_result):
+ """Returns "continue", "recheck", "force_build", or "abort"."""
+ while True:
+ print(
+ 'Got "{}" for pkg "{}", action: [C(ontinue), r(echeck), f(orce build),\
+ s(kip), b(ack) a(abort)]'.format(
+ state_result, pkg
+ )
+ )
+ user_input = sys.stdin.buffer.readline().decode().strip().lower()
+ if user_input == "c" or len(user_input) == 0:
+ return "continue"
+ elif user_input == "r":
+ return "recheck"
+ elif user_input == "f":
+ return "force_build"
+ elif user_input == "s":
+ return "skip"
+ elif user_input == "b":
+ return "back"
+ elif user_input == "a":
+ return "abort"
+ else:
+ log_print("Got invalid input")
+ continue
+
+
+def print_state_info_and_get_update_list(pkg_state):
+ to_update = []
+ log_print("package state:")
+ for (pkg_name, pkg_dict) in pkg_state.items():
+ if "state" in pkg_dict:
+ log_print(f" {pkg_name:40}: {pkg_dict['state']}")
+ if pkg_dict["state"] == "install":
+ to_update.append(pkg_name)
+ else:
+ log_print(f" {pkg_name:40}: not reached")
+ return to_update
+
+
+def test_gpg_passphrase(signing_gpg_dir, signing_key_fp, passphrase):
+ with tempfile.NamedTemporaryFile() as tempnf:
+ tempnf.write(b"Test file content")
+ tempnf.flush()
+ try:
+ subprocess.run(
+ [
+ "gpg",
+ "--batch",
+ "--passphrase-fd",
+ "0",
+ "--pinentry-mode",
+ "loopback",
+ "--default-key",
+ signing_key_fp,
+ "--detach-sign",
+ tempnf.name,
+ ],
+ check=True,
+ input=passphrase,
+ text=True,
+ env={"GNUPGHOME": signing_gpg_dir},
+ )
+ os.remove(tempnf.name + ".sig")
+ except subprocess.CalledProcessError:
+ log_print("ERROR: Failed to sign test file with gpg")
+ return False
+ log_print("Verified passphrase works by signing dummy test file")
+ return True
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser(description="Update AUR pkgs")
+ parser.add_argument(
+ "--config", help="Info and pkg(s) to update in a .toml config"
+ )
+ parser.add_argument(
+ "-p", "--pkg", action="append", help="Pkg(s) to update", metavar="pkg"
+ )
+ parser.add_argument(
+ "--no-skip",
+ action="append",
+ help="Pkg(s) to not skip if up to date",
+ metavar="noskip",
+ )
+ parser.add_argument(
+ "-e",
+ "--editor",
+ default="vim",
+ help="editor to use when viewing PKGBUILDs",
+ metavar="editor",
+ )
+ parser.add_argument("--chroot", help="Chroot to build in")
+ parser.add_argument("--pkg-dir", help="Destination for built pkgs")
+ parser.add_argument("--repo", help="repository tar file")
+ parser.add_argument("--gpg-dir", help="gpg home for checking signatures")
+ parser.add_argument("--logs-dir", help="dir to put logs")
+ parser.add_argument(
+ "--no-update", help="Do not update chroot", action="store_true"
+ )
+ parser.add_argument("--signing-gpg-dir", help="gpg home for signing key")
+ parser.add_argument(
+ "--signing-gpg-key-fp", help="gpg fingerprint for signing key"
+ )
+ parser.add_argument(
+ "--no-store",
+ action="store_true",
+ help="Don't sign built package and add to repo",
+ )
+ args = parser.parse_args()
+
+ if (
+ args.pkg
+ and not args.config
+ and (
+ not args.chroot
+ or not args.pkg_dir
+ or not args.repo
+ or not args.gpg_dir
+ or not args.logs_dir
+ )
+ ):
+ print(
+ "ERROR: --pkg requires also --chroot, --pkg_dir, --repo, --gpg_dir, and --logs_dir"
+ )
+ sys.exit(1)
+
+ pkg_state = {}
+ if args.pkg and not args.config:
+ for pkg in args.pkg:
+ pkg_state[pkg] = {}
+ pkg_state[pkg]["aur_deps"] = []
+ args_chroot = args.chroot
+ args_pkg_dir = args.pkg_dir
+ args_repo = args.repo
+ args_gpg_home = args.gpg_dir
+ args_logs_dir = args.logs_dir
+ if args_logs_dir is not None:
+ GLOBAL_LOG_FILE = args_logs_dir + "/update.py_logs"
+ log_print(
+ f"{datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d %H:%M %Z')}"
+ )
+ log_print(f"Set GLOBAL_LOG_FILE to {GLOBAL_LOG_FILE}")
+ args_signing_gpg_dir = args.signing_gpg_dir
+ args_signing_gpg_key_fp = args.signing_gpg_key_fp
+ if args_signing_gpg_key_fp is None:
+ log_print(
+ 'ERROR: Signing key fingerprint "signing_gpg_key_fp" not present in config'
+ )
+ sys.exit(1)
+ if args_signing_gpg_dir is not None and not args.no_store:
+ args_signing_gpg_pass = getpass.getpass("gpg signing key pass: ")
+ if not test_gpg_passphrase(
+ args_signing_gpg_dir,
+ args_signing_gpg_key_fp,
+ args_signing_gpg_pass,
+ ):
+ sys.exit(1)
+ elif args.config:
+ d = toml.load(args.config)
+ for entry in d["entry"]:
+ pkg_state[entry["name"]] = {}
+ if "aur_deps" in entry:
+ pkg_state[entry["name"]]["aur_deps"] = entry["aur_deps"]
+ else:
+ pkg_state[entry["name"]]["aur_deps"] = []
+ if "repo_path" in entry:
+ pkg_state[entry["name"]]["repo_path"] = entry["repo_path"]
+ if "pkg_name" in entry:
+ pkg_state[entry["name"]]["pkg_name"] = entry["pkg_name"]
+ else:
+ pkg_state[entry["name"]]["pkg_name"] = entry["name"]
+ if "ccache_dir" in entry:
+ pkg_state[entry["name"]]["ccache_dir"] = entry["ccache_dir"]
+ elif "sccache_dir" in entry:
+ pkg_state[entry["name"]]["sccache_dir"] = entry["sccache_dir"]
+ if "sccache_cache_size" in entry:
+ pkg_state[entry["name"]]["sccache_cache_size"] = entry[
+ "sccache_cache_size"
+ ]
+ else:
+ pkg_state[entry["name"]]["sccache_cache_size"] = "5G"
+ if "other_deps" in entry:
+ pkg_state[entry["name"]]["other_deps"] = entry["other_deps"]
+ else:
+ pkg_state[entry["name"]]["other_deps"] = []
+ if "skip_branch_up_to_date" in entry and not (
+ not args.no_skip is None and entry["name"] in args.no_skip
+ ):
+ pkg_state[entry["name"]]["skip_branch_up_to_date"] = True
+ else:
+ pkg_state[entry["name"]]["skip_branch_up_to_date"] = False
+ args_chroot = d["chroot"]
+ args_pkg_dir = d["pkg_dir"]
+ args_repo = d["repo"]
+ args_gpg_home = d["gpg_dir"]
+ args_logs_dir = d["logs_dir"]
+ if args_logs_dir is not None:
+ GLOBAL_LOG_FILE = args_logs_dir + "/update.py_logs"
+ log_print(
+ f"{datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d %H:%M %Z')}"
+ )
+ log_print(f"Set GLOBAL_LOG_FILE to {GLOBAL_LOG_FILE}")
+ if args.pkg:
+ to_keep = [args_pkg for args_pkg in args.pkg]
+ removal = []
+ for existing in pkg_state.keys():
+ if existing in to_keep:
+ pass
+ else:
+ removal.append(existing)
+ for to_remove in removal:
+ del pkg_state[to_remove]
+
+ if "signing_gpg_dir" in d and not args.no_store:
+ args_signing_gpg_dir = d["signing_gpg_dir"]
+ args_signing_gpg_key_fp = d["signing_gpg_key_fp"]
+ args_signing_gpg_pass = getpass.getpass("gpg signing key pass: ")
+ if not test_gpg_passphrase(
+ args_signing_gpg_dir,
+ args_signing_gpg_key_fp,
+ args_signing_gpg_pass,
+ ):
+ sys.exit(1)
+ else:
+ log_print('ERROR: At least "--config" or "--pkg" must be specified')
+ sys.exit(1)
+
+ os.putenv("CHROOT", os.path.realpath(args_chroot))
+ os.putenv("GNUPGHOME", os.path.realpath(args_gpg_home))
+ if not os.path.exists(args_logs_dir):
+ os.makedirs(args_logs_dir)
+ elif not os.path.isdir(args_logs_dir):
+ log_print(
+ 'ERROR: logs_dir "{}" must be a directory'.format(args_logs_dir)
+ )
+ sys.exit(1)
+ pkg_list = [temp_pkg_name for temp_pkg_name in pkg_state.keys()]
+ i = 0
+ while i < len(pkg_list):
+ going_back = False
+ if not ensure_pkg_dir_exists(pkg_list[i], pkg_state):
+ print_state_info_and_get_update_list(pkg_state)
+ sys.exit(1)
+ skip = False
+ if (
+ "repo_path" not in pkg_state[pkg_list[i]]
+ or pkg_state[pkg_list[i]]["repo_path"] != "NO_REPO"
+ ):
+ update_pkg_dir_count = 0
+ update_pkg_dir_success = False
+ while update_pkg_dir_count < 5:
+ (success, skip_on_same_ver) = update_pkg_dir(pkg_list[i], pkg_state)
+ if success:
+ update_pkg_dir_success = True
+ break
+ else:
+ time.sleep(1)
+ update_pkg_dir_count += 1
+ if not update_pkg_dir_success:
+ log_print('Failed to update pkg dir for "{}"', pkg_list[i])
+ print_state_info_and_get_update_list(pkg_state)
+ sys.exit(1)
+ if skip_on_same_ver:
+ check_pkg_version_result = check_pkg_version(pkg_list[i], pkg_state, args_repo, True)
+ if check_pkg_version_result != "install":
+ log_print(f'Pkg {pkg_list[i]} is up to date, skipping...')
+ pkg_state[pkg_list[i]]["state"] = "up to date"
+ i += 1
+ continue
+ else:
+ check_pkg_build_result = check_pkg_build(pkg_list[i], args.editor)
+ if check_pkg_build_result == "ok":
+ pass
+ elif check_pkg_build_result == "not_ok":
+ pkg_state[pkg_list[i]]["state"] = "skip"
+ i += 1
+ continue
+ elif check_pkg_build_result == "force_build":
+ pkg_state[pkg_list[i]]["state"] = "install"
+ i += 1
+ continue
+ elif check_pkg_build_result == "invalid":
+ continue
+ elif check_pkg_build_result == "back":
+ if i > 0:
+ i -= 1
+ continue
+ else: # check_pkg_build_result == "abort":
+ print_state_info_and_get_update_list(pkg_state)
+ sys.exit(1)
+ while True:
+ if skip_on_same_ver and check_pkg_version_result is not None:
+ state_result = check_pkg_version_result
+ else:
+ state_result = check_pkg_version(pkg_list[i], pkg_state, args_repo, False)
+ confirm_result_result = confirm_result(pkg_list[i], state_result)
+ if confirm_result_result == "continue":
+ pkg_state[pkg_list[i]]["state"] = state_result
+ break
+ elif confirm_result_result == "recheck":
+ check_pkg_version_result = None
+ continue
+ elif confirm_result_result == "force_build":
+ pkg_state[pkg_list[i]]["state"] = "install"
+ break
+ elif confirm_result_result == "skip":
+ pkg_state[pkg_list[i]]["state"] = "skip"
+ break
+ elif confirm_result_result == "back":
+ if i > 0:
+ i -= 1
+ going_back = True
+ break
+ else: # confirm_result_result == "abort"
+ print_state_info_and_get_update_list(pkg_state)
+ sys.exit(1)
+ if going_back:
+ pass
+ else:
+ i += 1
+
+ log_print("Showing current actions:")
+ pkgs_to_update = print_state_info_and_get_update_list(pkg_state)
+ if len(pkgs_to_update) > 0:
+ log_print("Continue? [Y/n]")
+ user_input = sys.stdin.buffer.readline().decode().strip().lower()
+ if user_input == "y" or len(user_input) == 0:
+ if args.no_update:
+ log_print("Updating (without updating chroot)...")
+ else:
+ log_print("Updating...")
+ update_pkg_list(
+ pkgs_to_update,
+ pkg_state,
+ os.path.realpath(args_chroot),
+ os.path.realpath(args_pkg_dir),
+ os.path.realpath(args_repo),
+ os.path.realpath(args_logs_dir),
+ args.no_update,
+ "" if args.no_store else args_signing_gpg_dir,
+ "" if args.no_store else args_signing_gpg_key_fp,
+ "" if args.no_store else args_signing_gpg_pass,
+ args.no_store,
+ )
+ else:
+ log_print("Canceled.")
+ else:
+ log_print("No packages to update, done.")
projects / AnotherAURHelper / commitdiff