Update updater script

.github/workflows/updater.py

@@ -1,49 +1,59 @@
 #!/usr/bin/env python3
+"""
+This script is meant to be run by GitHub Actions.
+It comes with a Github Action updater.yml to run this script periodically.
+
+Since each app is different, maintainers can adapt its contents to perform
+automatic actions when a new upstream release is detected.
+
+You need to enable the action by removing `if ${{ false }}` in updater.yml!
+"""
 
 import hashlib
 import json
 import logging
 import os
-import subprocess
+import re
+from subprocess import run, PIPE
 import textwrap
-from pathlib import Path
 from typing import List, Tuple, Any
 import requests
 from packaging import version
 
 logging.getLogger().setLevel(logging.INFO)
 
-# This script is meant to be run by GitHub Actions
+# ========================================================================== #
-# The YunoHost-Apps organisation offers a template Action to run this script periodically
+# Functions customizable by app maintainer
-# Since each app is different, maintainers can adapt its contents so as to perform
-# automatic actions when a new upstream release is detected.
-
-# Remove this exit command when you are ready to run this Action
-# exit(1)
-
-#=================================================
-# Fetching information
 
 def get_latest_version(repo: str) -> Tuple[version.Version, Any]:
-    api_url = repo.replace("https://github.com/", "https://api.github.com/repos/")
+    """May be customized by maintainers for other forges than Github"""
+    api_url = repo.replace("github.com", "api.github.com/repos")
+    # May use {api_url}/tags and release["name"] for tag-based upstream
     releases = requests.get(f"{api_url}/tags").json()
-    release_info = [
+    release_info = next(
         release for release in releases
         if "-rc" not in release["name"] and "REL" not in release["name"]
-    ][0]
+    )
     return version.Version(release_info["name"]), release_info
 
-def get_assets_of_release(repo: str, rel_info: Any) -> List[str]:
+def get_asset_urls_of_release(repo: str, release: Any) -> List[str]:
     """May be customized by maintainers for custom urls"""
-    rel = rel_info['name']
+    rel = release['name']
     short_rel = '.'.join(rel.split('.')[:2])
-    assets = [
+    return [
         f"https://releases.wikimedia.org/mediawiki/{short_rel}/mediawiki-{rel}.tar.gz"
     ]
-    return assets
 
-#=================================================
+def handle_asset(asset_url: str):
-# Download assets and compute filename / sha256sum
+    """This should be customized by the maintainer according to upstream"""
+    logging.info("Handling asset at %s", asset_url)
+    if asset_url.endswith(".tar.gz"):
+        write_src_file("app.src", asset_url, "tar.gz")
+    else:
+        logging.info("Asset ignored")
+
+# ========================================================================== #
+# Core generic code of the script
 
 def sha256sum_of_url(url: str) -> str:
     """Compute checksum without saving the file"""
@@ -52,56 +62,58 @@ def sha256sum_of_url(url: str) -> str:
         checksum.update(chunk)
     return checksum.hexdigest()
 
-# It has to be adapted in accordance with how the upstream releases look like.
+def write_src_file(name: str, asset_url: str, extension: str,
-def handle_asset(asset_url: str):
+                   extract: bool = True, subdir: bool = True) -> None:
-    """This should be customized by the maintainer"""
+    """Rewrite conf/app.src"""
-    logging.info("Handling asset at %s", asset_url)
+    logging.info("Writing %s...", name)
-    if asset_url.endswith(".tar.gz"):
-        src = "app.src"
-        extract = "true"
-    else:
-        logging.info("Asset ignored")
-        return
-    logging.info("Asset is for %s", src)
 
-    # Rewrite source file
+    with open(f"conf/{name}", "w", encoding="utf-8") as conf_file:
-    extension = "tar.gz" if asset_url.endswith(".tar.gz") else Path(asset_url).suffix[1:]
-    with open(f"conf/{src}", "w", encoding="utf-8") as conf_file:
         conf_file.write(textwrap.dedent(f"""\
             SOURCE_URL={asset_url}
             SOURCE_SUM={sha256sum_of_url(asset_url)}
             SOURCE_SUM_PRG=sha256sum
             SOURCE_FORMAT={extension}
-            SOURCE_IN_SUBDIR=true
+            SOURCE_IN_SUBDIR={str(subdir).lower()}
-            SOURCE_EXTRACT={extract}
+            SOURCE_EXTRACT={str(extract).lower()}
         """))
 
+def write_github_env(proceed: bool, new_version: str, branch: str):
+    """Those values will be used later in the workflow"""
+    if "GITHUB_ENV" not in os.environ:
+        logging.warning("GITHUB_ENV is not in the envvars, assuming not in CI")
+        return
+    with open(os.environ["GITHUB_ENV"], "w", encoding="utf-8") as github_env:
+        github_env.write(textwrap.dedent(f"""\
+            VERSION={new_version}
+            BRANCH={branch}
+            PROCEED={str(proceed).lower()}
+        """))
 
 def main():
-    with open(os.environ["GITHUB_ENV"], "w", encoding="utf-8") as github_env:
+    with open("manifest.json", "r", encoding="utf-8") as manifest_file:
-        github_env.write("PROCEED=false\n")
+        manifest = json.load(manifest_file)
-
-    with open("manifest.json", "r", encoding="utf-8") as file:
-        manifest = json.load(file)
     repo = manifest["upstream"]["code"]
 
     current_version = version.Version(manifest["version"].split("~")[0])
-    logging.info("Current version: %s", current_version)
     latest_version, release_info = get_latest_version(repo)
+    logging.info("Current version: %s", current_version)
     logging.info("Latest upstream version: %s", latest_version)
 
     # Proceed only if the retrieved version is greater than the current one
     if latest_version <= current_version:
         logging.warning("No new version available")
+        write_github_env(False, "", "")
         return
 
     # Proceed only if a PR for this new version does not already exist
-    command = ["git", "ls-remote", "--exit-code", "-h", repo, f"ci-auto-update-v${latest_version}"]
+    branch = f"ci-auto-update-v{latest_version}"
-    if subprocess.run(command, stderr=subprocess.DEVNULL, check=False).returncode == 0:
+    command = ["git", "ls-remote", "--exit-code", "-h", repo, branch]
+    if run(command, stderr=PIPE, stdout=PIPE, check=False).returncode == 0:
         logging.warning("A branch already exists for this update")
+        write_github_env(False, "", "")
         return
 
-    assets = get_assets_of_release(repo, release_info)
+    assets = get_asset_urls_of_release(repo, release_info)
     logging.info("%d available asset(s)", len(assets))
     for asset in assets:
         handle_asset(asset)
@@ -111,12 +123,7 @@ def main():
         json.dump(manifest, manifest_file, indent=4, ensure_ascii=False)
         manifest_file.write("\n")
 
-    with open(os.environ["GITHUB_ENV"], "w", encoding="utf-8") as github_env:
+    write_github_env(True, latest_version, branch)
-        github_env.write(textwrap.dedent(f"""\
-            VERSION={latest_version}
-            REPO={repo}
-            PROCEED=true
-        """))
 
 
 if __name__ == "__main__":