hosting/ynh-apps_tools
1
0
Fork 0
Code Activity

Improve new autoupdate mechanism, support latest_github_commit, be able to specify custom upstream to check, run in dry mode if passing an app path

Browse source
This commit is contained in:
Alexandre Aubin 2023-03-28 19:13:50 +02:00
parent 04f318cdf4
commit b688382b54
1 changed files with 91 additions and 54 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

117
autoupdate_app_sources/autoupdate_app_sources.py
View file

@ -7,16 +7,20 @@ import toml
import os import os
import glob import glob
from github import Github, InputGitAuthor STRATEGIES = ["latest_github_release", "latest_github_tag", "latest_github_commit"]
STRATEGIES = ["latest_github_release", "latest_github_tag"] if len(sys.argv) >= 2:
dry_run = True
else:
dry_run = False
GITHUB_LOGIN = open(os.path.dirname(__file__) + "/../../.github_login").read().strip() GITHUB_LOGIN = open(os.path.dirname(__file__) + "/../../.github_login").read().strip()
GITHUB_TOKEN = open(os.path.dirname(__file__) + "/../../.github_token").read().strip() GITHUB_TOKEN = open(os.path.dirname(__file__) + "/../../.github_token").read().strip()
GITHUB_EMAIL = open(os.path.dirname(__file__) + "/../../.github_email").read().strip() GITHUB_EMAIL = open(os.path.dirname(__file__) + "/../../.github_email").read().strip()
github = Github(GITHUB_TOKEN) from github import Github, InputGitAuthor
author = InputGitAuthor(GITHUB_LOGIN, GITHUB_EMAIL) github = Github(GITHUB_TOKEN)
author = InputGitAuthor(GITHUB_LOGIN, GITHUB_EMAIL)
def apps_to_run_auto_update_for(): def apps_to_run_auto_update_for():
@ -54,18 +58,23 @@ def apps_to_run_auto_update_for():
return out return out
def filter_and_get_latest_tag(tags): def filter_and_get_latest_tag(tags, app_id):
filter_keywords = ["start", "rc", "beta", "alpha"] filter_keywords = ["start", "rc", "beta", "alpha"]
tags = [t for t in tags if not any(keyword in t for keyword in filter_keywords)] tags = [t for t in tags if not any(keyword in t for keyword in filter_keywords)]
tag_dict = {}
for t in tags: for t in tags:
if not re.match(r"^v?[\d\.]*\d$", t): t_to_check = t
print(f"Ignoring tag {t}, doesn't look like a version number") if t.startswith(app_id + "-"):
tags = [t for t in tags if re.match(r"^v?[\d\.]*\d$", t)] t_to_check = t.split("-", 1)[-1]
tag_dict = {t: tag_to_int_tuple(t) for t in tags} if not re.match(r"^v?[\d\.]*\d$", t_to_check):
tags = sorted(tags, key=tag_dict.get) print(f"Ignoring tag {t_to_check}, doesn't look like a version number")
return tags[-1] else:
tag_dict[t] = tag_to_int_tuple(t_to_check)
tags = sorted(list(tag_dict.keys()), key=tag_dict.get)
return tags[-1], '.'.join([str(i) for i in tag_dict[tags[-1]]])
def tag_to_int_tuple(tag): def tag_to_int_tuple(tag):
@ -92,9 +101,13 @@ def sha256_of_remote_file(url):
class AppAutoUpdater: class AppAutoUpdater:
def __init__(self, app_id): def __init__(self, app_id):
# if not os.path.exists(app_path + "/manifest.toml"): if dry_run:
# raise Exception("manifest.toml doesnt exists?") if not os.path.exists(app_id + "/manifest.toml"):
raise Exception("manifest.toml doesnt exists?")
# app_id is in fact a path
manifest = toml.load(open(app_id + "/manifest.toml"))
else:
# We actually want to look at the manifest on the "testing" (or default) branch # We actually want to look at the manifest on the "testing" (or default) branch
self.repo = github.get_repo(f"Yunohost-Apps/{app_id}_ynh") self.repo = github.get_repo(f"Yunohost-Apps/{app_id}_ynh")
# Determine base branch, either `testing` or default branch # Determine base branch, either `testing` or default branch
@ -108,13 +121,14 @@ class AppAutoUpdater:
self.manifest_raw_sha = contents.sha self.manifest_raw_sha = contents.sha
manifest = toml.loads(self.manifest_raw) manifest = toml.loads(self.manifest_raw)
self.app_id = manifest["id"]
self.current_version = manifest["version"].split("~")[0] self.current_version = manifest["version"].split("~")[0]
self.sources = manifest.get("resources", {}).get("sources") self.sources = manifest.get("resources", {}).get("sources")
if not self.sources: if not self.sources:
raise Exception("There's no resources.sources in manifest.toml ?") raise Exception("There's no resources.sources in manifest.toml ?")
self.upstream = manifest.get("upstream", {}).get("code") self.main_upstream = manifest.get("upstream", {}).get("code")
def run(self): def run(self):
@ -136,13 +150,12 @@ class AppAutoUpdater:
print(f"Checking {source} ...") print(f"Checking {source} ...")
new_version, new_asset_urls = self.get_latest_version_and_asset( new_version, new_asset_urls = self.get_latest_version_and_asset(
strategy, asset, infos strategy, asset, infos, source
) )
if source == "main":
print(f"Current version in manifest: {self.current_version}") print(f"Current version in manifest: {self.current_version}")
print(f"Newest version on upstream: {new_version}") print(f"Newest version on upstream: {new_version}")
if source == "main":
if self.current_version == new_version: if self.current_version == new_version:
print( print(
f"Version is still {new_version}, no update required for {source}" f"Version is still {new_version}, no update required for {source}"
@ -171,7 +184,7 @@ class AppAutoUpdater:
"old_assets": infos, "old_assets": infos,
} }
if not todos: if dry_run or not todos:
return return
if "main" in todos: if "main" in todos:
@ -218,38 +231,40 @@ class AppAutoUpdater:
print("Created PR " + self.repo.full_name + " updated with PR #" + str(pr.id)) print("Created PR " + self.repo.full_name + " updated with PR #" + str(pr.id))
def get_latest_version_and_asset(self, strategy, asset, infos): def get_latest_version_and_asset(self, strategy, asset, infos, source):
upstream = infos.get("autoupdate", {}).get("upstream", self.main_upstream)
if "github" in strategy: if "github" in strategy:
assert self.upstream and self.upstream.startswith( assert upstream and upstream.startswith(
"https://github.com/" "https://github.com/"
), "When using strategy {strategy}, having a defined upstream code repo on github.com is required" ), f"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( upstream_repo = upstream.replace("https://github.com/", "").strip(
"/" "/"
) )
assert ( assert (
len(self.upstream_repo.split("/")) == 2 len(upstream_repo.split("/")) == 2
), "'{self.upstream}' doesn't seem to be a github repository ?" ), f"'{upstream}' doesn't seem to be a github repository ?"
if strategy == "latest_github_release": if strategy == "latest_github_release":
releases = self.github(f"repos/{self.upstream_repo}/releases") releases = self.github_api(f"repos/{upstream_repo}/releases")
tags = [ tags = [
release["tag_name"] release["tag_name"]
for release in releases for release in releases
if not release["draft"] and not release["prerelease"] if not release["draft"] and not release["prerelease"]
] ]
latest_version = filter_and_get_latest_tag(tags) latest_version_orig, latest_version = filter_and_get_latest_tag(tags, self.app_id)
if asset == "tarball": if asset == "tarball":
latest_tarball = ( latest_tarball = (
f"{self.upstream}/archive/refs/tags/{latest_version}.tar.gz" f"{upstream}/archive/refs/tags/{latest_version_orig}.tar.gz"
) )
return latest_version.strip("v"), latest_tarball return latest_version, latest_tarball
# FIXME # FIXME
else: else:
latest_release = [ latest_release = [
release release
for release in releases for release in releases
if release["tag_name"] == latest_version if release["tag_name"] == latest_version_orig
][0] ][0]
latest_assets = { latest_assets = {
a["name"]: a["browser_download_url"] a["name"]: a["browser_download_url"]
@ -270,7 +285,7 @@ class AppAutoUpdater:
raise Exception( raise Exception(
f"Too many assets matching regex '{asset}' for release {latest_version} : {matching_assets_urls}" f"Too many assets matching regex '{asset}' for release {latest_version} : {matching_assets_urls}"
) )
return latest_version.strip("v"), matching_assets_urls[0] return latest_version, matching_assets_urls[0]
elif isinstance(asset, dict): elif isinstance(asset, dict):
matching_assets_dicts = {} matching_assets_dicts = {}
for asset_name, asset_regex in asset.items(): for asset_name, asset_regex in asset.items():
@ -295,17 +310,35 @@ class AppAutoUpdater:
raise Exception( raise Exception(
"For the latest_github_tag strategy, only asset = 'tarball' is supported" "For the latest_github_tag strategy, only asset = 'tarball' is supported"
) )
tags = self.github(f"repos/{self.upstream_repo}/tags") tags = self.github_api(f"repos/{upstream_repo}/tags")
latest_version = filter_and_get_latest_tag([t["name"] for t in tags]) latest_version_orig, latest_version = filter_and_get_latest_tag([t["name"] for t in tags], self.app_id)
latest_tarball = ( latest_tarball = (
f"{self.upstream}/archive/refs/tags/{latest_version}.tar.gz" f"{upstream}/archive/refs/tags/{latest_version}.tar.gz"
) )
return latest_version.strip("v"), latest_tarball return latest_version, latest_tarball
elif strategy == "latest_github_commit":
if asset != "tarball":
raise Exception(
"For the latest_github_release strategy, only asset = 'tarball' is supported"
)
commits = self.github_api(f"repos/{upstream_repo}/commits")
latest_commit = commits[0]
latest_tarball = f"https://github.com/{upstream_repo}/archive/{latest_commit['sha']}.tar.gz"
# Let's have the version as something like "2023.01.23"
latest_version = latest_commit["commit"]["author"]["date"][:10].replace("-", ".")
return latest_version, latest_tarball
def github_api(self, uri):
if dry_run:
auth = None
else:
auth = (GITHUB_LOGIN, GITHUB_TOKEN)
def github(self, uri):
# print(f'https://api.github.com/{uri}')
r = requests.get( r = requests.get(
f"https://api.github.com/{uri}", auth=(GITHUB_LOGIN, GITHUB_TOKEN) f"https://api.github.com/{uri}", auth=None
) )
assert r.status_code == 200, r assert r.status_code == 200, r
return r.json() return r.json()
@ -362,5 +395,9 @@ def progressbar(it, prefix="", size=60, file=sys.stdout):
if __name__ == "__main__": if __name__ == "__main__":
if len(sys.argv) >= 2:
AppAutoUpdater(sys.argv[1]).run()
else:
for app in progressbar(apps_to_run_auto_update_for(), "Checking: ", 40): for app in progressbar(apps_to_run_auto_update_for(), "Checking: ", 40):
AppAutoUpdater(app).run() AppAutoUpdater(app).run()