import os import re import requests from itertools import groupby from github import Github from github.GithubException import GithubException repos_URL = { "stable": "netdata/netdata", "nightly": "netdata/netdata-nightlies" } GH_TOKEN = os.getenv("GH_TOKEN") if GH_TOKEN is None or GH_TOKEN != "": print("Token is not defined or empty, continuing with limitation on requests per sec towards Github API") def identify_channel(_version): nightly_pattern = r'v(\d+)\.(\d+)\.(\d+)-(\d+)-nightly' stable_pattern = r'v(\d+)\.(\d+)\.(\d+)' if re.match(nightly_pattern, _version): _channel = "nightly" _pattern = nightly_pattern elif re.match(stable_pattern, _version): _channel = "stable" _pattern = stable_pattern else: print("Invalid version format.") return None return _channel, _pattern def padded_version(item): key_value = '10000' for value in item[1:]: key_value += f'{value:05}' return int(key_value) def extract_version(title): if identify_channel(title): _, _pattern = identify_channel(title) try: match = re.match(_pattern, title) if match: return tuple(map(int, match.groups())) except Exception as e: print(f"Unexpected error: {e}") return None def get_release_path_and_filename(_version): nightly_pattern = r'v(\d+)\.(\d+)\.(\d+)-(\d+)-nightly' stable_pattern = r'v(\d+)\.(\d+)\.(\d+)' if match := re.match(nightly_pattern, _version): msb = match.group(1) _path = "nightly" _filename = f"v{msb}" elif match := re.match(stable_pattern, _version): msb = match.group(1) _path = "stable" _filename = f"v{msb}" else: print("Invalid version format.") exit(1) return (_path, _filename) def compare_version_with_remote(version): """ If the version = fun (version) you need to update the version in the remote. If the version remote doesn't exist, returns the version :param channel: any version of the agent :return: the greater from version and version remote. """ prefix = "https://packages.netdata.cloud/releases" path, filename = get_release_path_and_filename(version) remote_url = f"{prefix}/{path}/{filename}" response = requests.get(remote_url) if response.status_code == 200: version_remote = response.text.rstrip() version_components = extract_version(version) remote_version_components = extract_version(version_remote) absolute_version = padded_version(version_components) absolute_remote_version = padded_version(remote_version_components) if absolute_version > absolute_remote_version: print(f"Version in the remote: {version_remote}, is older than the current: {version}, I need to update") return (version) else: print(f"Version in the remote: {version_remote}, is newer than the current: {version}, no action needed") return (None) else: # Remote version not found print(f"Version in the remote not found, updating the predefined latest path with the version: {version}") return (version) def sort_and_grouby_major_agents_of_channel(channel): """ Fetches the GH API and read either netdata/netdata or netdata/netdata-nightlies repo. It fetches all of their releases implements a grouping by their major release number. Every k,v in this dictionary is in the form; "vX": [descending ordered list of Agents in this major release]. :param channel: "nightly" or "stable" :return: None or dict() with the Agents grouped by major version # (vX) """ try: G = Github(GH_TOKEN) repo = G.get_repo(repos_URL[channel]) releases = repo.get_releases() except GithubException as e: print(f"GitHub API request failed: {e}") return None except Exception as e: print(f"An unexpected error occurred: {e}") return None extracted_titles = [extract_version(item.title) for item in releases if extract_version(item.title) is not None] # Necessary sorting for implement the group by extracted_titles.sort(key=lambda x: x[0]) # Group titles by major version grouped_by_major = {major: list(group) for major, group in groupby(extracted_titles, key=lambda x: x[0])} sorted_grouped_by_major = {} for key, values in grouped_by_major.items(): sorted_values = sorted(values, key=padded_version, reverse=True) sorted_grouped_by_major[key] = sorted_values # Transform them in the correct form if channel == "stable": result_dict = {f"v{key}": [f"v{a}.{b}.{c}" for a, b, c in values] for key, values in sorted_grouped_by_major.items()} else: result_dict = {f"v{key}": [f"v{a}.{b}.{c}-{d}-nightly" for a, b, c, d in values] for key, values in sorted_grouped_by_major.items()} return result_dict