mirror of
https://github.com/aphrody-labs/msys2-devtools.git
synced 2026-10-14 04:19:12 +00:00
230 lines
7.7 KiB
Python
230 lines
7.7 KiB
Python
import sys
|
|
import argparse
|
|
import os
|
|
import json
|
|
from collections import OrderedDict
|
|
import hashlib
|
|
import time
|
|
import shlex
|
|
import subprocess
|
|
import gzip
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from functools import partial
|
|
|
|
from typing import List, Iterator, Tuple, Dict, Optional, Union, Collection, Sequence, Any
|
|
from .pkgbuild import get_extra_meta_for_pkgbuild
|
|
from .pkgextra import extra_to_pkgextra_entry
|
|
|
|
|
|
CacheEntry = Dict[str, Union[str, Collection[str]]]
|
|
CacheTuple = Tuple[str, CacheEntry]
|
|
Cache = Dict[str, CacheEntry]
|
|
|
|
|
|
def normalize_repo(repo: str) -> str:
|
|
if repo.endswith(".git"):
|
|
repo = repo.rsplit(".", 1)[0]
|
|
return repo
|
|
|
|
|
|
def normalize_path(path: str) -> str:
|
|
return path.replace("\\", "/")
|
|
|
|
|
|
def get_mingw_arch_list(msys2_root: str, dir: str, pkgbuild_path: str) -> List[str]:
|
|
assert not os.path.isabs(pkgbuild_path)
|
|
executable = os.path.join(msys2_root, 'usr', 'bin', 'bash.exe')
|
|
sub_commands = [
|
|
shlex.join(['source', pkgbuild_path]),
|
|
'! declare -p mingw_arch &>/dev/null',
|
|
'echo -n "$? ${mingw_arch[@]}"'
|
|
]
|
|
env = os.environ.copy()
|
|
env["CHERE_INVOKING"] = "1"
|
|
env["MSYSTEM"] = "MSYS"
|
|
env["MSYS2_PATH_TYPE"] = "minimal"
|
|
out = subprocess.check_output(
|
|
[executable, '-lc', ';'.join(sub_commands)], universal_newlines=True, env=env, cwd=dir)
|
|
first, *arch_list = out.strip().split()
|
|
list_exists = bool(int(first))
|
|
if not list_exists:
|
|
assert not arch_list
|
|
arch_list = ["mingw32", "mingw64", "ucrt64", "clang64", "clang32"]
|
|
return arch_list
|
|
|
|
|
|
def check_output_msys(msys2_root: str, args: Sequence[str], **kwargs: Any):
|
|
executable = os.path.join(msys2_root, 'usr', 'bin', 'bash.exe')
|
|
env = kwargs.pop("env", os.environ.copy())
|
|
env["CHERE_INVOKING"] = "1"
|
|
env["MSYSTEM"] = "MSYS"
|
|
env["MSYS2_PATH_TYPE"] = "minimal"
|
|
return subprocess.check_output(
|
|
[executable, '-lce'] + [shlex.join([str(a) for a in args])],
|
|
env=env, **kwargs)
|
|
|
|
|
|
def get_cache_key(pkgbuild_path: str) -> str:
|
|
pkgbuild_path = os.path.abspath(pkgbuild_path)
|
|
git_cwd = os.path.dirname(pkgbuild_path)
|
|
git_path = os.path.relpath(pkgbuild_path, git_cwd)
|
|
h = hashlib.new("SHA1")
|
|
|
|
with open(pkgbuild_path, "rb") as f:
|
|
h.update(f.read())
|
|
|
|
fileinfo = subprocess.check_output(
|
|
["git", "ls-files", "-s", "--full-name", git_path],
|
|
cwd=git_cwd).decode("utf-8").strip()
|
|
h.update(normalize_path(fileinfo).encode("utf-8"))
|
|
|
|
repo = subprocess.check_output(
|
|
["git", "ls-remote", "--get-url", "origin"],
|
|
cwd=git_cwd).decode("utf-8").strip()
|
|
repo = normalize_repo(repo)
|
|
h.update(repo.encode("utf-8"))
|
|
|
|
return h.hexdigest()
|
|
|
|
|
|
def get_srcinfo_for_pkgbuild(msys2_root: str, args: Tuple[str, str]) -> Optional[CacheTuple]:
|
|
pkgbuild_path, mode = args
|
|
pkgbuild_path = os.path.abspath(pkgbuild_path)
|
|
git_cwd = os.path.dirname(pkgbuild_path)
|
|
git_path = os.path.relpath(pkgbuild_path, git_cwd)
|
|
key = get_cache_key(pkgbuild_path)
|
|
|
|
print("Parsing %r" % pkgbuild_path)
|
|
try:
|
|
srcinfos = {}
|
|
|
|
if mode == "mingw":
|
|
for name in get_mingw_arch_list(msys2_root, git_cwd, git_path):
|
|
env = os.environ.copy()
|
|
env["MINGW_ARCH"] = name
|
|
srcinfo = check_output_msys(
|
|
msys2_root,
|
|
["/usr/bin/makepkg-mingw",
|
|
"--printsrcinfo", "-p", git_path],
|
|
cwd=git_cwd,
|
|
env=env).decode("utf-8")
|
|
assert srcinfo
|
|
srcinfos[name] = srcinfo
|
|
else:
|
|
srcinfo = check_output_msys(
|
|
msys2_root,
|
|
["/usr/bin/makepkg",
|
|
"--printsrcinfo", "-p", git_path],
|
|
cwd=git_cwd).decode("utf-8")
|
|
assert srcinfo
|
|
srcinfos["msys"] = srcinfo
|
|
|
|
repo = subprocess.check_output(
|
|
["git", "ls-remote", "--get-url", "origin"],
|
|
cwd=git_cwd).decode("utf-8").strip()
|
|
repo = normalize_repo(repo)
|
|
|
|
relpath = subprocess.check_output(
|
|
["git", "ls-files", "--full-name", git_path],
|
|
cwd=git_cwd).decode("utf-8").strip()
|
|
relpath = normalize_path(os.path.dirname(relpath))
|
|
|
|
date = subprocess.check_output(
|
|
["git", "log", "-1", "--format=%aI", git_path],
|
|
cwd=git_cwd).decode("utf-8").strip()
|
|
|
|
extra_meta = get_extra_meta_for_pkgbuild(msys2_root, pkgbuild_path)
|
|
|
|
meta = {"repo": repo, "path": relpath, "date": date, "srcinfo": srcinfos, "extra": extra_meta}
|
|
except subprocess.CalledProcessError as e:
|
|
print("ERROR: %s %s" % (pkgbuild_path, e.output.splitlines()))
|
|
return None
|
|
|
|
return (key, meta)
|
|
|
|
|
|
def iter_pkgbuild_paths(repo_path: str) -> Iterator[str]:
|
|
repo_path = os.path.abspath(repo_path)
|
|
print("Searching for PKGBUILD files in %s" % repo_path)
|
|
for base, dirs, files in os.walk(repo_path):
|
|
for f in files:
|
|
if f == "PKGBUILD":
|
|
# in case we find a PKGBUILD, don't go deeper
|
|
del dirs[:]
|
|
path = os.path.join(base, f)
|
|
yield path
|
|
|
|
|
|
def get_srcinfo_from_cache(args: Tuple[str, Cache]) -> Tuple[str, Optional[CacheTuple]]:
|
|
pkgbuild_path, cache = args
|
|
key = get_cache_key(pkgbuild_path)
|
|
if key in cache:
|
|
return (pkgbuild_path, (key, cache[key]))
|
|
else:
|
|
return (pkgbuild_path, None)
|
|
|
|
|
|
def iter_srcinfo(msys2_root: str, repo_path: str, mode: str, cache: Cache) -> Iterator[Optional[CacheTuple]]:
|
|
with ThreadPoolExecutor() as executor:
|
|
to_parse: List[Tuple[str, str]] = []
|
|
pool_iter = executor.map(
|
|
get_srcinfo_from_cache, ((p, cache) for p in iter_pkgbuild_paths(repo_path)))
|
|
for pkgbuild_path, srcinfo in pool_iter:
|
|
if srcinfo is not None:
|
|
yield srcinfo
|
|
else:
|
|
to_parse.append((pkgbuild_path, mode))
|
|
|
|
print("Parsing PKGBUILD files...")
|
|
for srcinfo in executor.map(partial(get_srcinfo_for_pkgbuild, msys2_root), to_parse):
|
|
yield srcinfo
|
|
|
|
|
|
def validate_srcinfo(entry: CacheEntry):
|
|
if "extra" in entry:
|
|
extra_to_pkgextra_entry(entry["extra"])
|
|
|
|
|
|
def main(argv: List[str]) -> None:
|
|
parser = argparse.ArgumentParser(description="Create SRCINFOs for all packages in a repo", allow_abbrev=False)
|
|
parser.add_argument('mode', choices=['msys', 'mingw'], help="The type of the repo")
|
|
parser.add_argument("msys2_root", help="The path to MSYS2")
|
|
parser.add_argument("repo_path", help="The path to GIT repo")
|
|
parser.add_argument("json_cache", help="The path to the json.gz file used to fetch/store the results")
|
|
parser.add_argument(
|
|
"--time-limit", action="store",
|
|
type=int, dest="time_limit", default=0,
|
|
help='time after which it will stop and save, 0 means no limit')
|
|
args = parser.parse_args(argv[1:])
|
|
|
|
t = time.monotonic()
|
|
|
|
srcinfo_path = os.path.abspath(args.json_cache)
|
|
cache: Cache = {}
|
|
try:
|
|
with open(srcinfo_path, "rb") as h:
|
|
cache = json.loads(gzip.decompress(h.read()))
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
srcinfos = []
|
|
for entry in iter_srcinfo(args.msys2_root, args.repo_path, args.mode, cache):
|
|
if entry is None:
|
|
continue
|
|
validate_srcinfo(entry[1])
|
|
srcinfos.append(entry)
|
|
# So we stop before CI times out
|
|
if args.time_limit and time.monotonic() - t > args.time_limit:
|
|
print("time limit reached, stopping")
|
|
break
|
|
|
|
srcinfos_dict = OrderedDict(sorted(srcinfos))
|
|
with open(srcinfo_path, "wb") as h:
|
|
h.write(gzip.compress(json.dumps(srcinfos_dict, indent=2).encode("utf-8")))
|
|
|
|
return None
|
|
|
|
|
|
def run() -> None:
|
|
return main(sys.argv)
|