mirror of
https://github.com/YunoHost/apps.git
synced 2024-09-03 20:06:07 +02:00
240 lines
10 KiB
Python
240 lines
10 KiB
Python
import time
|
|
import hashlib
|
|
import re
|
|
import sys
|
|
import requests
|
|
import toml
|
|
import os
|
|
|
|
from github import Github, InputGitAuthor
|
|
|
|
#from rich.traceback import install
|
|
#install(width=150, show_locals=True, locals_max_length=None, locals_max_string=None)
|
|
|
|
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()
|
|
GITHUB_EMAIL = open(os.path.dirname(__file__) + "/../../.github_email").read().strip()
|
|
|
|
github = Github(GITHUB_TOKEN)
|
|
author = InputGitAuthor(GITHUB_LOGIN, GITHUB_EMAIL)
|
|
|
|
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)
|
|
|
|
|
|
def sha256_of_remote_file(url):
|
|
print(f"Computing sha256sum for {url} ...")
|
|
try:
|
|
r = requests.get(url, stream=True)
|
|
m = hashlib.sha256()
|
|
for data in r.iter_content(8192):
|
|
m.update(data)
|
|
return m.hexdigest()
|
|
except Exception as e:
|
|
print(f"Failed to compute sha256 for {url} : {e}")
|
|
return None
|
|
|
|
|
|
class AppAutoUpdater():
|
|
|
|
def __init__(self, app_id):
|
|
|
|
#if not os.path.exists(app_path + "/manifest.toml"):
|
|
# raise Exception("manifest.toml doesnt exists?")
|
|
|
|
# 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")
|
|
# Determine base branch, either `testing` or default branch
|
|
try:
|
|
self.base_branch = self.repo.get_branch("testing").name
|
|
except:
|
|
self.base_branch = self.repo.default_branch
|
|
|
|
contents = self.repo.get_contents("manifest.toml", ref=self.base_branch)
|
|
self.manifest_raw = contents.decoded_content.decode()
|
|
self.manifest_raw_sha = contents.sha
|
|
manifest = toml.loads(self.manifest_raw)
|
|
|
|
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):
|
|
|
|
todos = {}
|
|
|
|
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} ...")
|
|
|
|
new_version, new_asset_urls = self.get_latest_version_and_asset(strategy, asset, infos)
|
|
|
|
print(f"Current version in manifest: {self.current_version}")
|
|
print(f"Newest version on upstream: {new_version}")
|
|
|
|
if source == "main":
|
|
if self.current_version == new_version:
|
|
print(f"Version is still {new_version}, no update required for {source}")
|
|
continue
|
|
else:
|
|
print(f"Update needed for {source}")
|
|
todos[source] = {"new_asset_urls": new_asset_urls, "old_assets": infos, "new_version": new_version}
|
|
else:
|
|
if isinstance(new_asset_urls, str) and infos["url"] == new_asset_urls:
|
|
print(f"URL is still up to date for asset {source}")
|
|
continue
|
|
elif isinstance(new_asset_urls, dict) and new_asset_urls == {k: infos[k]["url"] for k in new_asset_urls.keys()}:
|
|
print(f"URLs are still up to date for asset {source}")
|
|
continue
|
|
else:
|
|
print(f"Update needed for {source}")
|
|
todos[source] = {"new_asset_urls": new_asset_urls, "old_assets": infos}
|
|
|
|
if not todos:
|
|
return
|
|
|
|
if "main" in todos:
|
|
new_version = todos["main"]["new_version"]
|
|
message = f"Upgrade to v{new_version}"
|
|
new_branch = f"ci-auto-update-{new_version}"
|
|
else:
|
|
message = "Upgrade sources"
|
|
new_branch = "ci-auto-update-sources"
|
|
|
|
try:
|
|
# Get the commit base for the new branch, and create it
|
|
commit_sha = self.repo.get_branch(self.base_branch).commit.sha
|
|
self.repo.create_git_ref(ref=f"refs/heads/{new_branch}", sha=commit_sha)
|
|
except:
|
|
pass
|
|
|
|
manifest_new = self.manifest_raw
|
|
for source, infos in todos.items():
|
|
manifest_new = self.replace_version_and_asset_in_manifest(manifest_new, infos.get("new_version"), infos["new_asset_urls"], infos["old_assets"], is_main=source == "main")
|
|
|
|
self.repo.update_file("manifest.toml",
|
|
message=message,
|
|
content=manifest_new,
|
|
sha=self.manifest_raw_sha,
|
|
branch=new_branch,
|
|
author=author)
|
|
|
|
# Wait a bit to preserve the API rate limit
|
|
time.sleep(1.5)
|
|
|
|
# Open the PR
|
|
pr = self.repo.create_pull(title=message, body=message, head=new_branch, base=self.base_branch)
|
|
|
|
print("Created PR " + self.repo.full_name + " updated with PR #" + str(pr.id))
|
|
|
|
|
|
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()
|
|
|
|
def replace_version_and_asset_in_manifest(self, content, new_version, new_assets_urls, current_assets, is_main):
|
|
|
|
if isinstance(new_assets_urls, str):
|
|
sha256 = sha256_of_remote_file(new_assets_urls)
|
|
elif isinstance(new_assets_urls, dict):
|
|
sha256 = {url: sha256_of_remote_file(url) for url in new_assets_urls.values()}
|
|
|
|
if is_main:
|
|
def repl(m):
|
|
return m.group(1) + new_version + m.group(3)
|
|
content = re.sub(r"(\s*version\s*=\s*[\"\'])([\d\.]+)(\~ynh\d+[\"\'])", repl, content)
|
|
if isinstance(new_assets_urls, str):
|
|
content = content.replace(current_assets["url"], new_assets_urls)
|
|
content = content.replace(current_assets["sha256"], sha256)
|
|
elif isinstance(new_assets_urls, dict):
|
|
for key, url in new_assets_urls.items():
|
|
content = content.replace(current_assets[key]["url"], url)
|
|
content = content.replace(current_assets[key]["sha256"], sha256[url])
|
|
|
|
return content
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
AppAutoUpdater(sys.argv[1]).run()
|