gtn-to-usegalaxy-tools.py

#!/usr/bin/env python3
# Quick and dirty script to update usegalaxy-tools with missing tools fron GTN trainings. Note that this installs the
# newest versions of things, not necessarily the version specified in the training json.
import argparse
import os
import pathlib
import string
import subprocess
import sys

import requests
import yaml


def section_id_chr(c):
    return (c if c in string.ascii_letters + string.digits else '_').lower()


def section_label_to_id(label):
    return ''.join(map(section_id_chr, label))


def tools_on_server(server):
    r = requests.get(f"https://{server}/api/tools?in_panel=false")
    return r.json()


def tool_ids(tools):
    return [t["id"] for t in tools]


def tool_repo_owner(tool_id):
    tool_split = tool_id.split("/")
    owner = tool_split[2]
    repo = tool_split[3]
    return (owner, repo)


parser = argparse.ArgumentParser()
parser.add_argument("--usegalaxy-tools", help="Path to usegalaxy-tools repo", default=os.getcwd(), type=pathlib.Path)
parser.add_argument("tutorial_json", help="tutorial.json URL to parse")
args = parser.parse_args()


tutorial_r = requests.get(args.tutorial_json)
tutorial = tutorial_r.json()
tutorial_tools = tutorial["tools"]


org_tool_list = tools_on_server("usegalaxy.org")
eu_tool_list = None
au_tool_list = None
org_tools = tool_ids(org_tool_list)


update_or_install = []
for tool in tutorial_tools:
    if tool in org_tools:
        print(f"✔️ {tool}")
    else:
        print(f"x {tool}")
        update_or_install.append(tool)

install = []
toolset = args.usegalaxy_tools / "usegalaxy.org"
for tool in update_or_install:

    for fname in os.listdir(toolset):
        owner, repo = tool_repo_owner(tool)
        if fname.endswith('.yml'):
            tool_yaml = yaml.safe_load(open(toolset / fname))
            found = False
            for repo_dict in tool_yaml["tools"]:
                if repo_dict["owner"] == owner and repo_dict["name"] == repo:
                    found = True
                    break
            if found:
                break
    if found:
        print(f"{owner}/{repo} found in {str(fname)}")
        print("Running update-tool.py")
        subprocess.check_call([sys.executable, "scripts/update-tool.py", "--owner", owner, "--name", repo, "usegalaxy.org/" + fname], cwd=args.usegalaxy_tools)
    else:
        print(f"{owner}/{repo} not found in usegalaxy-tools")
        install.append(tool)


if install:
    section_name = None
    for tool in install:
        owner, repo = tool_repo_owner(tool)
        if not eu_tool_list:
            print("Fetching EU tool list")
            eu_tool_list = tools_on_server("usegalaxy.eu")
        for tool_dict in eu_tool_list:
            if tool == tool_dict["id"]:
                section_name = tool_dict["panel_section_name"]
                print(f"Found {owner}/{repo} on usegalaxy.eu in section {section_name}")
                break
        if not section_name:
            if not au_tool_list:
                print("Fetching AU tool list")
                au_tool_list = tools_on_server("usegalaxy.org.au")
            for tool_dict in au_tool_list:
                if tool == tool_dict["id"]:
                    section_name = tool_dict["panel_section_name"]
                    print(f"Found {owner}/{repo} on usegalaxy.org.au in section {section_name}")
                    break
        if not section_name:
            print(f"ERROR: Cannot find tool on other usegalaxy.* servers: {owner}/{repo}")
            continue
        section_id = section_label_to_id(section_name)
        fname = section_id + ".yml"
        fpath = toolset / fname
        if not os.path.exists(fpath):
            print(f"ERROR: Section YAML does not exist: {fpath}")
            continue
        print(f"Writing {owner}/{repo} to {fpath}")
        tool_yaml = yaml.safe_load(open(fpath))
        tool_yaml["tools"].append({"owner": owner, "name": repo})
        yaml.dump(tool_yaml, open(fpath, "w"))
        print(f"Running fix-lockfile.py")
        subprocess.check_call([sys.executable, "scripts/fix-lockfile.py", "usegalaxy.org/" + fname], cwd=args.usegalaxy_tools)
        print(f"Running update-tool.py --without")
        subprocess.check_call([sys.executable, "scripts/update-tool.py", "--without", "usegalaxy.org/" + fname], cwd=args.usegalaxy_tools)