From d589511cb2825d153ca4598ddde880a61f0bb2ff Mon Sep 17 00:00:00 2001 From: Alexandre Aubin Date: Mon, 13 Mar 2023 17:40:35 +0100 Subject: [PATCH] POC for new declarative app source auto-update mechanism --- .../autoupdate_app_sources.py | 125 ++++++++++++++++++ 1 file changed, 125 insertions(+) create mode 100644 autoupdate_app_sources/autoupdate_app_sources.py diff --git a/autoupdate_app_sources/autoupdate_app_sources.py b/autoupdate_app_sources/autoupdate_app_sources.py new file mode 100644 index 0000000..9333fe8 --- /dev/null +++ b/autoupdate_app_sources/autoupdate_app_sources.py @@ -0,0 +1,125 @@ +import re +import sys +import requests +import toml +import os + +STRATEGIES = ["latest_github_release", "latest_github_tag"] + +GITHUB_LOGIN = open(os.path.dirname(__file__) + "/../../.github_login").read().strip() +GITHUB_TOKEN = open(os.path.dirname(__file__) + "/../../.github_token").read().strip() + + +def filter_and_get_latest_tag(tags): + filter_keywords = ["start", "rc", "beta", "alpha"] + tags = [t for t in tags if not any(keyword in t for keyword in filter_keywords)] + + for t in tags: + if not re.match(r"^v?[\d\.]*\d$", t): + print(f"Ignoring tag {t}, doesn't look like a version number") + tags = [t for t in tags if re.match(r"^v?[\d\.]*\d$", t)] + + tag_dict = {t: tag_to_int_tuple(t) for t in tags} + tags = sorted(tags, key=tag_dict.get) + return tags[-1] + + +def tag_to_int_tuple(tag): + + tag = tag.strip("v") + int_tuple = tag.split(".") + assert all(i.isdigit() for i in int_tuple), f"Cant convert {tag} to int tuple :/" + return tuple(int(i) for i in int_tuple) + + +class AppAutoUpdater(): + + def __init__(self, app_path): + + if not os.path.exists(app_path + "/manifest.toml"): + raise Exception("manifest.toml doesnt exists?") + + manifest = toml.load(open(app_path + "/manifest.toml")) + + self.current_version = manifest["version"].split("~")[0] + self.sources = manifest.get("resources", {}).get("sources") + + if not self.sources: + raise Exception("There's no resources.sources in manifest.toml ?") + + self.upstream = manifest.get("upstream", {}).get("code") + + def run(self): + + for source, infos in self.sources.items(): + + if "autoupdate" not in infos: + continue + + strategy = infos.get("autoupdate", {}).get("strategy") + if strategy not in STRATEGIES: + raise Exception(f"Unknown strategy to autoupdate {source}, expected one of {STRATEGIES}, got {strategy}") + + asset = infos.get("autoupdate", {}).get("asset", "tarball") + + print(f"Checking {source} ...") + + version, assets = self.get_latest_version_and_asset(strategy, asset, infos) + + print(f"Current version in manifest: {self.current_version}") + print(f"Newest version on upstream: {version}") + print(assets) + + def get_latest_version_and_asset(self, strategy, asset, infos): + + if "github" in strategy: + assert self.upstream and self.upstream.startswith("https://github.com/"), "When using strategy {strategy}, having a defined upstream code repo on github.com is required" + self.upstream_repo = self.upstream.replace("https://github.com/", "").strip("/") + assert len(self.upstream_repo.split("/")) == 2, "'{self.upstream}' doesn't seem to be a github repository ?" + + if strategy == "latest_github_release": + releases = self.github(f"repos/{self.upstream_repo}/releases") + tags = [release["tag_name"] for release in releases if not release["draft"] and not release["prerelease"]] + latest_version = filter_and_get_latest_tag(tags) + if asset == "tarball": + latest_tarball = f"{self.upstream}/archive/refs/tags/{latest_version}.tar.gz" + return latest_version.strip("v"), latest_tarball + # FIXME + else: + latest_release = [release for release in releases if release["tag_name"] == latest_version][0] + latest_assets = {a["name"]: a["browser_download_url"] for a in latest_release["assets"] if not a["name"].endswith(".md5")} + if isinstance(asset, str): + matching_assets_urls = [url for name, url in latest_assets.items() if re.match(asset, name)] + if not matching_assets_urls: + raise Exception(f"No assets matching regex '{asset}' for release {latest_version} among {list(latest_assets.keys())}") + elif len(matching_assets_urls) > 1: + raise Exception(f"Too many assets matching regex '{asset}' for release {latest_version} : {matching_assets_urls}") + return latest_version.strip("v"), matching_assets_urls[0] + elif isinstance(asset, dict): + matching_assets_dicts = {} + for asset_name, asset_regex in asset.items(): + matching_assets_urls = [url for name, url in latest_assets.items() if re.match(asset_regex, name)] + if not matching_assets_urls: + raise Exception(f"No assets matching regex '{asset}' for release {latest_version} among {list(latest_assets.keys())}") + elif len(matching_assets_urls) > 1: + raise Exception(f"Too many assets matching regex '{asset}' for release {latest_version} : {matching_assets_urls}") + matching_assets_dicts[asset_name] = matching_assets_urls[0] + return latest_version.strip("v"), matching_assets_dicts + + elif strategy == "latest_github_tag": + if asset != "tarball": + raise Exception("For the latest_github_tag strategy, only asset = 'tarball' is supported") + tags = self.github(f"repos/{self.upstream_repo}/tags") + latest_version = filter_and_get_latest_tag([t["name"] for t in tags]) + latest_tarball = f"{self.upstream}/archive/refs/tags/{latest_version}.tar.gz" + return latest_version.strip("v"), latest_tarball + + def github(self, uri): + #print(f'https://api.github.com/{uri}') + r = requests.get(f'https://api.github.com/{uri}', auth=(GITHUB_LOGIN, GITHUB_TOKEN)) + assert r.status_code == 200, r + return r.json() + + +if __name__ == "__main__": + AppAutoUpdater(sys.argv[1]).run()