# Copyright John N. Laliberte # Copyright Mart Raudsepp # LICENSE - GPL2 # vim: set sts=4 sw=4 et tw=0 : import requests try: from requests_futures.sessions import FuturesSession parallel_requests = True except: print("requests-futures not found for parallel fetching - will fallback to slower one-by-one version retrieval for latest version") parallel_requests = False import package_module, clioptions_module MAX_WORKERS = 10 DEBUG = False # TODO: Figure out some better handling of mappings together with package_module # TODO: package_module has made the reverse mapping of what we need, because we consume the names from release_packages # TODO: So this reverses it back again for the modules we care for this until a proper fix name_mapping = { "cantarell": "cantarell-fonts", "libgfbgraph": "gfbgraph", "nm-applet": "network-manager-applet", "networkmanager": "NetworkManager", "gnome-themes-standard": "gnome-themes-extra", } # Turns GNOME versions (40.beta) into the PMS version format (40_beta) def normalize_version(version: str) -> str: if version.endswith(('.rc', '.beta', '.alpha')): i = version.rindex('.') return f"{version[:i]}_{version[i + 1:]}" return version class GNOME: def __init__(self, nextrev=False): options = clioptions_module.Options() args = options.get_arguments() self.nextrev = nextrev; self.major_release = ".".join( args.release_number.split(".")[:2] ) self.full_release = args.release_number # gnome_release is the major release number the development is leading up to or is part of. E.g, 2.21.5 will turn to 2.22 and 2.22.2 becomes 2.22 # This is used in latest versions list fetching URL construction gnome_release_list = args.release_number.split(".") if int(gnome_release_list[0]) < 40 and int(gnome_release_list[1]) % 2 != 0: gnome_release_list[1] = str(int(gnome_release_list[1]) + 1) self.gnome_release = ".".join(gnome_release_list[:2]) if parallel_requests: self.http = FuturesSession(max_workers=MAX_WORKERS) else: self.http = requests.session() self.url_base = "https://download.gnome.org/" self.release_versions_file_path = self.url_base + 'teams/releng/' def generate_data_from_versions_markup(self, url): data = self.http.get(url) if parallel_requests: data = data.result() if not data: raise ValueError("Couldn't open %s" % url) ret = [] for line in data.text.splitlines(): components = line.strip().split(':') # Skip invalid lines if len(components) != 4 or len(components[2]) == 0: continue pkg = package_module.Package(components[1] + "-" + normalize_version(components[2]), suite=components[0]) if (pkg.name and pkg.version): ret.append(pkg) else: print("Warning: Ignoring package %s because parsing of its name or version string '%s' failed" % (components[1], components[2])) return ret def generate_data_individual(self, release_packages): ret = [] # First query all results; if parallel_requests==True, this will run in parallel for pkg in release_packages: name = pkg.name.split('/')[-1] if name in name_mapping: name = name_mapping[name] pkg.requests_result = self.http.get(self.url_base + '/sources/' + name + '/cache.json') # And now handle the results - this is a separate loop for parallel fetch support for pkg in release_packages: name = pkg.name.split('/')[-1] if name in name_mapping: name = name_mapping[name] # pkg.requests_result is the resulting Response if parallel_requests else Future that we need to call result() on to wait/retrieve the response data = pkg.requests_result.result() if parallel_requests else pkg.requests_result if not data: print("Warning: Unable to read cache.json for %s" % pkg.name) continue data = data.json() if data[0] != 4: print("Warning: unknown cache.json version for package %s" % name) continue major_minor = pkg.major_minor new_versioning = False if major_minor not in data[3]: major_minor = pkg.major new_versioning = True if major_minor not in data[3]: print("Warning: can't find latest version for %s-%s" % (name, pkg.major_minor)) continue if not new_versioning and pkg.major != "0" and pkg.minor.isdigit() and int(pkg.minor) % 2 and "%s.%d" % (pkg.major, int(pkg.minor)+1) in data[3]: major_minor = "%s.%d" % (pkg.major, int(pkg.minor)+1) latest = False # Some modules contain more than LATEST-IS-* for some reason, so we need to iterate and find the correct item instead of [0] (even though it is firsy always, but lets be future-proof) for tarball in data[3][major_minor]: if tarball.startswith('LATEST-IS-'): latest = tarball[10:] # len('LATEST-IS-') == 10 break if not latest: print("Warning: couldn't find latest version from within cache.json[3]") continue latest_pkg = package_module.Package(name + "-" + normalize_version(latest)) if pkg.name and pkg.version: ret.append(latest_pkg) else: print("Warning: Ignoring package %s because parsing of its name or version string '%s' failed" % (name, latest)) continue return ret def generate_data_release(self): return self.generate_data_from_versions_markup(self.release_versions_file_path + self.full_release + '/versions') import portage_module def compare_packages(release_packages, latest_packages, packages_in_portage): # we care about 5 cases # 1. portage version is less than the release version. (RED) # 2. portage version is equal to the release version, but less than the latest version. (LIGHT GREEN) # 3. portage version is equal to the release version and the latest version. (GREEN) # 4. portage version is greater than the release version (GREEN) # 5. package does not exist in portage (GREY) # again, we choose release_packages as the enumerator for the package names # since it will have ALL packages ( for example, if we used portage_packages, we # might miss the packages that do not exist in portage ) status_packages = [] for package in release_packages: color = None release_package = package latest_package = portage_module.findpackage(package.name, latest_packages) if latest_package is None: print("Warning: latest packages set does not have an entry for %s, using release set version %s." % (package.name, package.version)) latest_package = package portage_package = portage_module.findpackage(package.name, packages_in_portage) if portage_package is None: status = package_module.Status.NotFound # we need to create a phony package since findpackage # returns None portage_package = package_module.Package(package.name) elif portage_module.best_version_test(portage_package.name_plus_version, \ release_package.name_plus_version) == 2: status = package_module.Status.NeedUpdate elif portage_module.best_version_test(portage_package.name_plus_version, \ latest_package.name_plus_version) == 2: status = package_module.Status.NewerVersion else: status = package_module.Status.Compliant #if portage_package is not None: # FIXME: Reports release version as latest version to not have to deal with this right now if latest_package is None: print("No latest version known for %s, FIXME!" % release_package.name) latest_package = release_package if DEBUG: print("package: " + str(release_package.name) + " | pp: " + str(portage_package.version) + " | rp: " + str(release_package.version) + " | lp: " + str(latest_package.version) + " | status: " + str(status)) status_packages.append(package_module.PackageStatus(release_package.name, str(portage_package.version), str(release_package.version), str(latest_package.version), status, suite=release_package.suite)) return status_packages