ynh-apps_tools/update_app_levels/update_app_levels.py

#!/usr/bin/env python3
"""
Update app catalog: commit, and create a merge request
"""

import argparse
import json
import logging
import tempfile
import textwrap
import time
from collections import OrderedDict
from typing import Any

from pathlib import Path
import jinja2
import requests
import toml
from git import Repo

# APPS_REPO = "YunoHost/apps"
APPS_REPO = "Salamandar/apps"

CI_RESULTS_URL = "https://ci-apps.yunohost.org/ci/api/results"

REPO_APPS_ROOT = Path(Repo(__file__, search_parent_directories=True).working_dir)

VERBOSE = False


def github_token() -> str | None:
    github_token_path = REPO_APPS_ROOT.parent / ".github_token"
    if github_token_path.exists():
        return github_token_path.open("r", encoding="utf-8").read().strip()
    return None


def get_ci_results() -> dict[str, dict[str, Any]]:
    return requests.get(CI_RESULTS_URL, timeout=10).json()


def ci_result_is_outdated(result) -> bool:
    # 3600 * 24 * 60 = ~2 months
    return (int(time.time()) - result.get("timestamp", 0)) > 3600 * 24 * 60


def update_catalog(catalog, ci_results) -> dict:
    """
    Actually change the catalog data
    """
    # Re-sort the catalog keys / subkeys
    for app, infos in catalog.items():
        catalog[app] = OrderedDict(sorted(infos.items()))
    catalog = OrderedDict(sorted(catalog.items()))

    def app_level(app):
        if app not in ci_results:
            return 0
        if ci_result_is_outdated(ci_results[app]):
            return 0
        return ci_results[app]["level"]

    for app, info in catalog.items():
        info["level"] = app_level(app)

    return catalog


def list_changes(catalog, ci_results) -> dict[str, list[tuple[str, int, int]]]:
    """
    Lists changes for a pull request
    """

    changes = {
        "major_regressions": [],
        "minor_regressions": [],
        "improvements": [],
        "outdated": [],
        "missing": [],
    }

    for app, infos in catalog.items():
        if infos.get("state") != "working":
            continue

        if app not in ci_results:
            changes["missing"].append(app)
            continue

        if ci_result_is_outdated(ci_results[app]):
            changes["outdated"].append(app)
            continue

        ci_level = ci_results[app]["level"]
        current_level = infos.get("level")

        if ci_level == current_level:
            continue

        if current_level is None or ci_level > current_level:
            changes["improvements"].append((app, current_level, ci_level))
            continue

        if ci_level < current_level:
            if ci_level <= 4 < current_level:
                changes["major_regressions"].append((app, current_level, ci_level))
            else:
                changes["minor_regressions"].append((app, current_level, ci_level))

    return changes


def pretty_changes(changes: dict[str, list[tuple[str, int, int]]]) -> str:
    pr_body_template = textwrap.dedent("""
        {%- if changes["major_regressions"] %}
        ### Major regressions 😭
        {% for app in changes["major_regressions"] %}
        - [ ] [{{app.0}}: {{app.1}} → {{app.2}}](https://ci-apps.yunohost.org/ci/apps/{{app.0}}/latestjob)
        {%- endfor %}
        {% endif %}
        {%- if changes["minor_regressions"] %}
        ### Minor regressions 😬
        {% for app in changes["minor_regressions"] %}
        - [ ] [{{app.0}}: {{app.1}} → {{app.2}}](https://ci-apps.yunohost.org/ci/apps/{{app.0}}/latestjob)
        {%- endfor %}
        {% endif %}
        {%- if changes["improvements"] %}
        ### Improvements 🥳
        {% for app in changes["improvements"] %}
        - [{{app.0}}: {{app.1}} → {{app.2}}](https://ci-apps.yunohost.org/ci/apps/{{app.0}}/latestjob)
        {%- endfor %}
        {% endif %}
        {%- if changes["missing"] %}
        ### Missing 🫠
        {% for app in changes["missing"] %}
        - [{{app}} (See latest job if it exists)](https://ci-apps.yunohost.org/ci/apps/{{app.0}}/latestjob)
        {%- endfor %}
        {% endif %}
        {%- if changes["outdated"] %}
        ### Outdated ⏰
        {% for app in changes["outdated"] %}
        - [ ] [{{app}} (See latest job if it exists)](https://ci-apps.yunohost.org/ci/apps/{{app.0}}/latestjob)
        {%- endfor %}
        {% endif %}
    """)

    return jinja2.Environment().from_string(pr_body_template).render(changes=changes)


def make_pull_request(pr_body: str) -> None:
    pr_data = {
        "title": "Update app levels according to CI results",
        "body": pr_body,
        "head": "update_app_levels",
        "base": "master"
    }

    with requests.Session() as s:
        s.headers.update({"Authorization": f"token {github_token()}"})
        response = s.post(f"https://api.github.com/repos/{APPS_REPO}/pulls", json=pr_data)

        if response.status_code == 422:
            response = s.get(f"https://api.github.com/repos/{APPS_REPO}/pulls", data={"head": "update_app_levels"})
            response.raise_for_status()
            pr_number = response.json()[0]["number"]

            # head can't be updated
            del pr_data["head"]
            response = s.patch(f"https://api.github.com/repos/{APPS_REPO}/pulls/{pr_number}", json=pr_data)
            response.raise_for_status()
            existing_url = response.json()["html_url"]
            logging.warning(f"An existing Pull Request has been updated at {existing_url} !")
        else:
            response.raise_for_status()

            new_url = response.json()["html_url"]
            logging.info(f"Opened a Pull Request at {new_url} !")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--commit", action=argparse.BooleanOptionalAction, default=True)
    parser.add_argument("--pr", action=argparse.BooleanOptionalAction, default=True)
    parser.add_argument("-v", "--verbose", action=argparse.BooleanOptionalAction)
    args = parser.parse_args()

    logging.getLogger().setLevel(logging.INFO)
    global VERBOSE
    if args.verbose:
        VERBOSE = True
        logging.getLogger().setLevel(logging.DEBUG)

    with tempfile.TemporaryDirectory(prefix="update_app_levels_") as tmpdir:
        logging.info("Cloning the repository...")
        apps_repo = Repo.clone_from(f"git@github.com:{APPS_REPO}", to_path=tmpdir)

        # Load the app catalog and filter out the non-working ones
        catalog = toml.load((Path(apps_repo.working_tree_dir) / "apps.toml").open("r", encoding="utf-8"))

        new_branch = apps_repo.create_head("update_app_levels", apps_repo.refs.master)
        apps_repo.head.reference = new_branch

        logging.info("Retrieving the CI results...")
        ci_results = get_ci_results()

        # Now compute changes, then update the catalog
        changes = list_changes(catalog, ci_results)
        pr_body = pretty_changes(changes)
        catalog = update_catalog(catalog, ci_results)

        # Save the new catalog
        updated_catalog = toml.dumps(catalog)
        updated_catalog = updated_catalog.replace(",]", " ]")
        (Path(apps_repo.working_tree_dir) / "apps.toml").open("w", encoding="utf-8").write(updated_catalog)

        if args.commit:
            logging.info("Committing and pushing the new catalog...")
            apps_repo.index.add("apps.toml")
            apps_repo.index.commit("Update app levels according to CI results")
            apps_repo.remote().push(force=True)

        if VERBOSE:
            print(pr_body)

        if args.pr:
            logging.info("Opening a pull request...")
            make_pull_request(pr_body)


if __name__ == "__main__":
    main()