Clone of https://github.com/NixOS/nixpkgs.git (to stress-test knotserver)
at devShellTools-shell 151 lines 5.3 kB view raw
1#! /usr/bin/env nix-shell 2#! nix-shell -i python3 -p python3 python3.pkgs.packaging python3.pkgs.requests python3.pkgs.xmltodict 3import json 4import pathlib 5import logging 6import requests 7import subprocess 8import sys 9from urllib.error import HTTPError 10import urllib.request 11import xmltodict 12from packaging import version 13 14updates_url = "https://www.jetbrains.com/updates/updates.xml" 15current_path = pathlib.Path(__file__).parent 16versions_file_path = current_path.joinpath("versions.json").resolve() 17fromVersions = {} 18toVersions = {} 19 20logging.basicConfig(stream=sys.stdout, level=logging.DEBUG) 21 22 23def one_or_more(x): 24 return x if isinstance(x, list) else [x] 25 26 27def download_channels(): 28 logging.info("Checking for updates from %s", updates_url) 29 updates_response = requests.get(updates_url) 30 updates_response.raise_for_status() 31 root = xmltodict.parse(updates_response.text) 32 products = root["products"]["product"] 33 return { 34 channel["@name"]: channel 35 for product in products 36 if "channel" in product 37 for channel in one_or_more(product["channel"]) 38 } 39 40 41def build_version(build): 42 build_number = build["@fullNumber"] if "@fullNumber" in build else build["@number"] 43 return version.parse(build_number) 44 45 46def latest_build(channel): 47 builds = one_or_more(channel["build"]) 48 latest = max(builds, key=build_version) 49 return latest 50 51 52def download_sha256(url): 53 url = f"{url}.sha256" 54 download_response = requests.get(url) 55 download_response.raise_for_status() 56 return download_response.content.decode('UTF-8').split(' ')[0] 57 58 59channels = download_channels() 60 61 62def get_url(template, version_or_build_number, version_number): 63 release = [str(n) for n in version.parse(version_number).release] 64 for k in range(len(release), 0, -1): 65 s = ".".join(release[0:k]) 66 url = template.format(version=version_or_build_number, versionMajorMinor=s) 67 try: 68 if urllib.request.urlopen(url).getcode() == 200: 69 return url 70 except HTTPError: 71 pass 72 return None 73 74 75def update_product(name, product): 76 update_channel = product["update-channel"] 77 logging.info("Updating %s", name) 78 channel = channels.get(update_channel) 79 if channel is None: 80 logging.error("Failed to find channel %s.", update_channel) 81 logging.error("Check that the update-channel in %s matches the name in %s", versions_file_path, updates_url) 82 else: 83 try: 84 build = latest_build(channel) 85 new_version = build["@version"] 86 new_build_number = "" 87 if "@fullNumber" not in build: 88 new_build_number = build["@number"] 89 else: 90 new_build_number = build["@fullNumber"] 91 if "EAP" not in channel["@name"]: 92 version_or_build_number = new_version 93 else: 94 version_or_build_number = new_build_number 95 version_number = new_version.split(' ')[0] 96 download_url = get_url(product["url-template"], version_or_build_number, version_number) 97 if not download_url: 98 raise Exception(f"No valid url for {name} version {version_or_build_number}") 99 product["url"] = download_url 100 if "sha256" not in product or product.get("build_number") != new_build_number: 101 fromVersions[name] = product["version"] 102 toVersions[name] = new_version 103 logging.info("Found a newer version %s with build number %s.", new_version, new_build_number) 104 product["version"] = new_version 105 product["build_number"] = new_build_number 106 product["sha256"] = download_sha256(download_url) 107 else: 108 logging.info("Already at the latest version %s with build number %s.", new_version, new_build_number) 109 except Exception as e: 110 logging.exception("Update failed:", exc_info=e) 111 logging.warning("Skipping %s due to the above error.", name) 112 logging.warning("It may be out-of-date. Fix the error and rerun.") 113 114 115def update_products(products): 116 for name, product in products.items(): 117 update_product(name, product) 118 119 120with open(versions_file_path, "r") as versions_file: 121 versions = json.load(versions_file) 122 123for products in versions.values(): 124 update_products(products) 125 126with open(versions_file_path, "w") as versions_file: 127 json.dump(versions, versions_file, indent=2) 128 versions_file.write("\n") 129 130if len(toVersions) == 0: 131 # No Updates found 132 sys.exit(0) 133 134if len(toVersions) == 1: 135 commitMessage = "" 136else: 137 lowestVersion = min(fromVersions.values()) 138 highestVersion = max(toVersions.values()) 139 commitMessage = f"jetbrains: {lowestVersion} -> {highestVersion}" 140 commitMessage += "\n\n" 141 142for name in toVersions.keys(): 143 commitMessage += f"jetbrains.{name}: {fromVersions[name]} -> {toVersions[name]}\n" 144 145# Commit the result 146logging.info("#### Committing changes... ####") 147subprocess.run(['git', 'commit', f'-m{commitMessage}', '--', f'{versions_file_path}'], check=True) 148 149logging.info("#### Updating plugins ####") 150plugin_script = current_path.joinpath("../plugins/update_plugins.py").resolve() 151subprocess.call(plugin_script)