natefoo · July 9, 2025 15:05
diff --git a/localetest.py b/localetest.py
 #!/usr/bin/env python3
 import argparse
 import os.path
 import json
 import re
 import shutil
 import subprocess
 import sys
 import urllib.parse
 from collections import defaultdict

 import requests


 DEFAULT_GALAXY_URL = "https://usegalaxy.org"
 RESOLVER_ENDPOINT = "/api/container_resolvers/toolbox"
 RESPONSE_FILE = "response.json"
 RESULTS_FILE = "results.json"
 APPTAINER_BIN = shutil.which("apptainer")


 def parse_args(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument("-k", "--api-key", help="Galaxy API key")
    parser.add_argument("-g", "--galaxy-url", default=DEFAULT_GALAXY_URL, help="Galaxy URL")
    parser.add_argument("-r", "--raw", action="store_true", help="Output raw Tool IDs instead of TPV regex")
    parser.add_argument("--skip-docker", action="store_true", help="Skip explicit docker containers")
    return parser.parse_args()


 def query_resolvers(galaxy_url, api_key):
    response = []
    if os.path.exists(RESPONSE_FILE):
        print(f"Using cached response, remove file for fresh request: {RESPONSE_FILE}", file=sys.stderr)
        with open(RESPONSE_FILE, "r") as fh:
            response = json.load(fh)
    else:
        url = urllib.parse.urljoin(galaxy_url, RESOLVER_ENDPOINT)
        print(f"Querying resolvers from: {url}", file=sys.stderr)
        r = requests.get(url, params={"key": api_key})
        r.raise_for_status()
        response = r.json()
        with open(RESPONSE_FILE, "w") as fh:
            json.dump(response, fh)
    return response


 def parse_response(response):
    container_map = {}
    for item in response:
        if item["status"]["model_class"] != "ContainerDependency":
            continue
        tool_id = item["tool_id"]
        container_id = item["status"]["container_description"]["identifier"]
        if container_id not in container_map:
            container_map[container_id] = []
        if tool_id not in container_map[container_id]:
            container_map[container_id].append(tool_id)
    return container_map


 def test_containers(container_map, skip_docker=None):
    results = {}
    if os.path.exists(RESULTS_FILE):
        print(f"Using cached results, remove file for fresh test: {RESULTS_FILE}", file=sys.stderr)
        with open(RESULTS_FILE, "r") as fh:
            results = json.load(fh)
    else:
        for container_id, tool_ids in container_map.items():
            if skip_docker and container_id.startswith("docker://"):
                print(f"Skipping explicit docker container: {container_id}", file=sys.stderr)
                continue
            for lc_all in ("C.UTF-8", "C"):
                if test_container(container_id, lc_all=lc_all):
                    if lc_all == "C":
                        print(f"WARNING: setlocale failed on LC_ALL=C: {container_id}", file=sys.stderr)
                    else:
                        results[container_id] = tool_ids
        with open(RESULTS_FILE, "w") as fh:
            json.dump(results, fh)
    return results


 def test_container(container_id, lc_all="C.UTF-8"):
    cmd = [APPTAINER_BIN, "exec", "--contain", "--cleanenv", "--no-mount=tmp", container_id, "/bin/bash", "-c", "true"]
    env = {"APPTAINERENV_LC_ALL": lc_all}
    try:
        p = subprocess.run(cmd, env=env, stderr=subprocess.PIPE, text=True)
        for line in p.stderr.splitlines():
            if "cannot change locale" in line:
                print(f"{lc_all}: {container_id}: {line}")
                return True
    except subprocess.CalledProcessError as exc:
        print(f"WARNING: APPTAINERENV_LC_ALL={lc_all} {' '.join(cmd)}: {str(exc)}", file=sys.stderr)
    return False


 def make_raw_list(results):
    for container_id in sorted(results.keys()):
        tool_ids = results[container_id]
        for tool_id in sorted(tool_ids):
            print(f"{container_id}\t{tool_id}")


 def make_tpv_list(results, raw=None):
    tools = set()
    for tool_ids in results.values():
        tools.update(set(tool_ids))
    tools = sorted(list(tools))
    for regex in regex_tool_ids(tools):
        print(regex)


 def regex_tool_ids(tool_ids):
    ddtree = lambda: defaultdict(ddtree)
    tree = ddtree()
    for tool_id in tool_ids:
        if '/' not in tool_id:
            print(tool_id)
            continue
        split_id = tool_id.split('/')
        tree[split_id[0]][split_id[1]][split_id[2]][split_id[3]][split_id[4]][split_id[5]]
    rval = []
    for ts, ts_d in tree.items():
        for r, r_d in ts_d.items():
            for owner, owner_d in r_d.items():
                for repo, repo_d in owner_d.items():
                    if len(repo_d) > 1:
                        s = [ts, r, owner, repo, os.path.commonprefix(list(repo_d.keys())) + "[^/]*"]
                        re = "/".join(s)
                        versions = unique_versions(repo_d.values())
                        for version in versions:
                            rval.append(f"{re}/{version}")
                    else:
                        for tool, tool_d in repo_d.items():
                            s = [ts, r, owner, repo, tool]
                            re = "/".join(s)
                            versions = unique_versions([tool_d])
                            for version in versions:
                                rval.append(f"{re}/{version}")
    return rval


 def unique_versions(tool_ds):
    # takes a list of tool_d
    all_versions = set()
    #for tool, tool_d in repo_d.items():
    for tool_d in tool_ds:
        for versions in list(tool_d.keys()):
            all_versions.add(versions)
    common_prefix = os.path.commonprefix(list(all_versions))
    majors = sorted(list(set([i.split(".")[0] for i in all_versions])))
    rval = []
    if len(all_versions) == 1:
        # there's only one version across all tools
        rval.append(re.escape(all_versions.pop()))
    elif common_prefix:
        # there's a common version prefix across all tools
        rval.append(re.escape(common_prefix) + ".*")
    else:
        # there's no common version prefix across all tools, so
        # add all single digit versions prior to the current inside a [] group
        rval.append("[" + "".join([m for m in majors[:-1] if len(m) == 1]) + "]\\..*")
        # add all double+ digit versions to the prior as separate entries
        for m in [m for m in majors[:-1] if len(m) > 1]:
            rval.append(m + "\\..*")
        # add all single digit minor versions of the current version inside a [] group
        minors = sorted(list(set([i.split(".")[1] for i in all_versions if i[0] == majors[-1]])))
        rval.append(majors[-1] + "\\.[" + "".join([m for m in minors if len(m) == 1]) + "]\\..*")
        # add all double+ digit minor versions of the current version as separate entries
        for m in [m for m in minors if len(m) > 1]:
            rval.append(majors[-1] + "\\." + m + "\\..*")
    return rval


 def main(argv=None):
    args = parse_args(argv)
    response = query_resolvers(args.galaxy_url, args.api_key)
    container_map = parse_response(response)
    results = test_containers(container_map, args.skip_docker)
    if args.raw:
        make_raw_list(results)
    else:
        make_tpv_list(results)


 if __name__ == "__main__":
    main(sys.argv)
diff --git a/no-utf8-container-list.tsv b/no-utf8-container-list.tsv
diff --git a/post-rebuild-1-list.tsv b/post-rebuild-1-list.tsv
	#!/usr/bin/env python3
	import argparse
	import os.path
	import json
	import re
	import shutil
	import subprocess
	import sys
	import urllib.parse
	from collections import defaultdict

	import requests


	DEFAULT_GALAXY_URL = "https://usegalaxy.org"
	RESOLVER_ENDPOINT = "/api/container_resolvers/toolbox"
	RESPONSE_FILE = "response.json"
	RESULTS_FILE = "results.json"
	APPTAINER_BIN = shutil.which("apptainer")


	def parse_args(argv=None):
	parser = argparse.ArgumentParser()
	parser.add_argument("-k", "--api-key", help="Galaxy API key")
	parser.add_argument("-g", "--galaxy-url", default=DEFAULT_GALAXY_URL, help="Galaxy URL")
	parser.add_argument("-r", "--raw", action="store_true", help="Output raw Tool IDs instead of TPV regex")
	parser.add_argument("--skip-docker", action="store_true", help="Skip explicit docker containers")
	return parser.parse_args()


	def query_resolvers(galaxy_url, api_key):
	response = []
	if os.path.exists(RESPONSE_FILE):
	print(f"Using cached response, remove file for fresh request: {RESPONSE_FILE}", file=sys.stderr)
	with open(RESPONSE_FILE, "r") as fh:
	response = json.load(fh)
	else:
	url = urllib.parse.urljoin(galaxy_url, RESOLVER_ENDPOINT)
	print(f"Querying resolvers from: {url}", file=sys.stderr)
	r = requests.get(url, params={"key": api_key})
	r.raise_for_status()
	response = r.json()
	with open(RESPONSE_FILE, "w") as fh:
	json.dump(response, fh)
	return response


	def parse_response(response):
	container_map = {}
	for item in response:
	if item["status"]["model_class"] != "ContainerDependency":
	continue
	tool_id = item["tool_id"]
	container_id = item["status"]["container_description"]["identifier"]
	if container_id not in container_map:
	container_map[container_id] = []
	if tool_id not in container_map[container_id]:
	container_map[container_id].append(tool_id)
	return container_map


	def test_containers(container_map, skip_docker=None):
	results = {}
	if os.path.exists(RESULTS_FILE):
	print(f"Using cached results, remove file for fresh test: {RESULTS_FILE}", file=sys.stderr)
	with open(RESULTS_FILE, "r") as fh:
	results = json.load(fh)
	else:
	for container_id, tool_ids in container_map.items():
	if skip_docker and container_id.startswith("docker://"):
	print(f"Skipping explicit docker container: {container_id}", file=sys.stderr)
	continue
	for lc_all in ("C.UTF-8", "C"):
	if test_container(container_id, lc_all=lc_all):
	if lc_all == "C":
	print(f"WARNING: setlocale failed on LC_ALL=C: {container_id}", file=sys.stderr)
	else:
	results[container_id] = tool_ids
	with open(RESULTS_FILE, "w") as fh:
	json.dump(results, fh)
	return results


	def test_container(container_id, lc_all="C.UTF-8"):
	cmd = [APPTAINER_BIN, "exec", "--contain", "--cleanenv", "--no-mount=tmp", container_id, "/bin/bash", "-c", "true"]
	env = {"APPTAINERENV_LC_ALL": lc_all}
	try:
	p = subprocess.run(cmd, env=env, stderr=subprocess.PIPE, text=True)
	for line in p.stderr.splitlines():
	if "cannot change locale" in line:
	print(f"{lc_all}: {container_id}: {line}")
	return True
	except subprocess.CalledProcessError as exc:
	print(f"WARNING: APPTAINERENV_LC_ALL={lc_all} {' '.join(cmd)}: {str(exc)}", file=sys.stderr)
	return False


	def make_raw_list(results):
	for container_id in sorted(results.keys()):
	tool_ids = results[container_id]
	for tool_id in sorted(tool_ids):
	print(f"{container_id}\t{tool_id}")


	def make_tpv_list(results, raw=None):
	tools = set()
	for tool_ids in results.values():
	tools.update(set(tool_ids))
	tools = sorted(list(tools))
	for regex in regex_tool_ids(tools):
	print(regex)


	def regex_tool_ids(tool_ids):
	ddtree = lambda: defaultdict(ddtree)
	tree = ddtree()
	for tool_id in tool_ids:
	if '/' not in tool_id:
	print(tool_id)
	continue
	split_id = tool_id.split('/')
	tree[split_id[0]][split_id[1]][split_id[2]][split_id[3]][split_id[4]][split_id[5]]
	rval = []
	for ts, ts_d in tree.items():
	for r, r_d in ts_d.items():
	for owner, owner_d in r_d.items():
	for repo, repo_d in owner_d.items():
	if len(repo_d) > 1:
	s = [ts, r, owner, repo, os.path.commonprefix(list(repo_d.keys())) + "[^/]*"]
	re = "/".join(s)
	versions = unique_versions(repo_d.values())
	for version in versions:
	rval.append(f"{re}/{version}")
	else:
	for tool, tool_d in repo_d.items():
	s = [ts, r, owner, repo, tool]
	re = "/".join(s)
	versions = unique_versions([tool_d])
	for version in versions:
	rval.append(f"{re}/{version}")
	return rval


	def unique_versions(tool_ds):
	# takes a list of tool_d
	all_versions = set()
	#for tool, tool_d in repo_d.items():
	for tool_d in tool_ds:
	for versions in list(tool_d.keys()):
	all_versions.add(versions)
	common_prefix = os.path.commonprefix(list(all_versions))
	majors = sorted(list(set([i.split(".")[0] for i in all_versions])))
	rval = []
	if len(all_versions) == 1:
	# there's only one version across all tools
	rval.append(re.escape(all_versions.pop()))
	elif common_prefix:
	# there's a common version prefix across all tools
	rval.append(re.escape(common_prefix) + ".*")
	else:
	# there's no common version prefix across all tools, so
	# add all single digit versions prior to the current inside a [] group
	rval.append("[" + "".join([m for m in majors[:-1] if len(m) == 1]) + "]\\..*")
	# add all double+ digit versions to the prior as separate entries
	for m in [m for m in majors[:-1] if len(m) > 1]:
	rval.append(m + "\\..*")
	# add all single digit minor versions of the current version inside a [] group
	minors = sorted(list(set([i.split(".")[1] for i in all_versions if i[0] == majors[-1]])))
	rval.append(majors[-1] + "\\.[" + "".join([m for m in minors if len(m) == 1]) + "]\\..*")
	# add all double+ digit minor versions of the current version as separate entries
	for m in [m for m in minors if len(m) > 1]:
	rval.append(majors[-1] + "\\." + m + "\\..*")
	return rval


	def main(argv=None):
	args = parse_args(argv)
	response = query_resolvers(args.galaxy_url, args.api_key)
	container_map = parse_response(response)
	results = test_containers(container_map, args.skip_docker)
	if args.raw:
	make_raw_list(results)
	else:
	make_tpv_list(results)


	if __name__ == "__main__":
	main(sys.argv)