#!/usr/bin/env python3 import argparse import json import logging import re from pathlib import Path from typing import ( Dict, Generator, Iterator, List, Optional, Set, Tuple, Union, ) from aqt.exceptions import ArchiveConnectionError, ArchiveDownloadError from aqt.helper import Settings, setup_logging from aqt.metadata import ArchiveId, MetadataFactory, Versions def is_blacklisted_tool(tool_name: str) -> bool: for prefix in ("tools_qt3dstudio_",): if tool_name.startswith(prefix): return True for suffix in ("_preview", "_early_access"): if tool_name.endswith(suffix): return True return False def iter_archive_ids( *, categories: Iterator[str] = ArchiveId.CATEGORIES, hosts: Iterator[str] = ArchiveId.HOSTS, targets: Optional[Iterator[str]] = None, add_extensions: bool = False, ) -> Generator[ArchiveId, None, None]: def iter_extensions() -> Generator[str, None, None]: if add_extensions: if cat == "qt6" and target == "android": yield from ("x86_64", "x86", "armv7", "arm64_v8a") return elif cat == "qt5" and target == "desktop": yield from ("wasm", "") return yield "" for cat in categories: for host in sorted(hosts): use_targets = targets if use_targets is None: use_targets = ArchiveId.TARGETS_FOR_HOST[host] for target in use_targets: if target == "winrt" and cat == "qt6": # there is no qt6 for winrt continue for ext in iter_extensions(): yield ArchiveId(cat, host, target, ext) def iter_arches() -> Generator[dict, None, None]: logger.info("Fetching arches") archive_ids = list(iter_archive_ids(categories=("qt5", "qt6"), add_extensions=True)) for archive_id in tqdm(archive_ids): versions = ( ("latest",) if archive_id.category == "qt6" else ("latest", "5.13.2", "5.9.9") ) for version in versions: if version == "5.9.9" and archive_id.extension == "wasm": continue for arch_name in MetadataFactory( archive_id, architectures_ver=version ).getList(): yield { "os_name": archive_id.host, "target": archive_id.target, "arch": arch_name, } def iter_tool_variants() -> Generator[dict, None, None]: for archive_id in iter_archive_ids(categories=("tools",)): logger.info("Fetching tool variants for {}".format(archive_id)) for tool_name in tqdm(sorted(MetadataFactory(archive_id).getList())): if is_blacklisted_tool(tool_name): continue for tool_variant in MetadataFactory( archive_id, tool_name=tool_name ).getList(): yield { "os_name": archive_id.host, "target": archive_id.target, "tool_name": tool_name, "arch": tool_variant, } def iter_qt_minor_groups( host: str = "linux", target: str = "desktop" ) -> Generator[Tuple[int, int], None, None]: for cat in ( "qt5", "qt6", ): versions: Versions = MetadataFactory(ArchiveId(cat, host, target)).getList() for minor_group in versions: v = minor_group[0] yield v.major, v.minor def iter_modules_for_qt_minor_groups( host: str = "linux", target: str = "desktop" ) -> Generator[Dict, None, None]: logger.info("Fetching qt modules for {}/{}".format(host, target)) for major, minor in tqdm(list(iter_qt_minor_groups(host, target))): cat = f"qt{major}" yield { "qt_version": f"{major}.{minor}", "modules": MetadataFactory( ArchiveId(cat, host, target), modules_ver=f"{major}.{minor}.0" ).getList(), } def list_qt_versions(host: str = "linux", target: str = "desktop") -> List[str]: all_versions = list() for cat in ( "qt5", "qt6", ): versions: Versions = MetadataFactory(ArchiveId(cat, host, target)).getList() for minor_group in versions: all_versions.extend([str(ver) for ver in minor_group]) return all_versions def merge_records(arch_records) -> List[Dict]: all_records: List[Dict] = [] hashes = set() for record in arch_records: _hash = record["os_name"], record["target"], record["arch"] if _hash not in hashes: all_records.append(record) hashes.add(_hash) for sorting_key in ("arch", "target", "os_name"): all_records = sorted(all_records, key=lambda d: d[sorting_key]) return all_records def generate_combos(new_archive: List[str]): return { "qt": merge_records(iter_arches()), "tools": list(iter_tool_variants()), "modules": list(iter_modules_for_qt_minor_groups()), "versions": list_qt_versions(), "new_archive": new_archive, } def pretty_print_combos(combos: Dict[str, Union[List[Dict], List[str]]]) -> str: """ Attempts to mimic the formatting of the existing combinations.json. """ def fmt_dict_entry(entry: Dict, depth: int) -> str: return '{}{{"os_name": {:<10} "target": {:<10} {}"arch": "{}"}}'.format( " " * depth, f'"{entry["os_name"]}",', f'"{entry["target"]}",', ( f'"tool_name": "{entry["tool_name"]}", ' if "tool_name" in entry.keys() else "" ), entry["arch"], ) def span_multiline(line: str, max_width: int, depth: int) -> str: window = (0, max_width) indent = " " * (depth + 1) while len(line) - window[0] > max_width: break_loc = line.rfind(" ", window[0], window[1]) line = line[:break_loc] + "\n" + indent + line[break_loc + 1 :] window = (break_loc + len(indent), break_loc + len(indent) + max_width) return line def fmt_module_entry(entry: Dict, depth: int = 0) -> str: line = '{}{{"qt_version": "{}", "modules": [{}]}}'.format( " " * depth, entry["qt_version"], ", ".join([f'"{s}"' for s in entry["modules"]]), ) return span_multiline(line, 120, depth) def fmt_version_list(entry: List[str], depth: int) -> str: assert isinstance(entry, list) minor_pattern = re.compile(r"^\d+\.(\d+)(\.\d+)?") def iter_minor_versions(): if len(entry) == 0: return begin_index = 0 current_minor_ver = int(minor_pattern.match(entry[begin_index]).group(1)) for i, ver in enumerate(entry): minor = int(minor_pattern.match(ver).group(1)) if minor != current_minor_ver: yield entry[begin_index:i] begin_index = i current_minor_ver = minor yield entry[begin_index:] joiner = ",\n" + " " * depth line = joiner.join( [ ", ".join([f'"{ver}"' for ver in minor_group]) for minor_group in iter_minor_versions() ] ) return line root_element_strings = [ f'"{key}": [\n' + ",\n".join([item_formatter(item, depth=1) for item in combos[key]]) + "\n]" for key, item_formatter in ( ("qt", fmt_dict_entry), ("tools", fmt_dict_entry), ("modules", fmt_module_entry), ) ] + [ f'"{key}": [\n ' + fmt_version_list(combos[key], depth=1) + "\n]" for key in ("versions", "new_archive") ] return "[{" + ", ".join(root_element_strings) + "}]" def compare_combos( actual_combos: Dict[str, Union[List[str], List[Dict]]], expected_combos: Dict[str, Union[List[str], List[Dict]]], actual_name: str, expect_name: str, ) -> bool: # list_of_str_keys: the values attached to these keys are List[str] list_of_str_keys = "versions", "new_archive" has_difference = False # Don't compare data pulled from previous file skipped_keys = ("new_archive",) def compare_modules_entry(actual_mod_item: Dict, expect_mod_item: Dict) -> bool: """Return True if difference detected. Print description of difference.""" version = actual_mod_item["qt_version"] actual_modules, expect_modules = set(actual_mod_item["modules"]), set( expect_mod_item["modules"] ) mods_missing_from_actual = expect_modules - actual_modules mods_missing_from_expect = actual_modules - expect_modules if mods_missing_from_actual: logger.info( f"{actual_name}['modules'] for Qt {version} is missing {mods_missing_from_actual}" ) if mods_missing_from_expect: logger.info( f"{expect_name}['modules'] for Qt {version} is missing {mods_missing_from_expect}" ) return bool(mods_missing_from_actual) or bool(mods_missing_from_expect) def to_set(a_list: Union[List[str], List[Dict]]) -> Set: if len(a_list) == 0: return set() if isinstance(a_list[0], str): return set(a_list) assert isinstance(a_list[0], Dict) return set([str(a_dict) for a_dict in a_list]) def report_difference( superset: Set, subset: Set, subset_name: str, key: str ) -> bool: """Return True if difference detected. Print description of difference.""" missing_from_superset = sorted(superset - subset) if not missing_from_superset: return False logger.info(f"{subset_name}['{key}'] is missing these entries:") if key in list_of_str_keys: logger.info(format(missing_from_superset)) return True for el in missing_from_superset: logger.info(format(el)) return True for root_key in actual_combos.keys(): if root_key in skipped_keys: continue logger.info(f"\nComparing {root_key}:\n{'-' * 40}") if root_key == "modules": for actual_row, expect_row in zip( actual_combos[root_key], expected_combos[root_key] ): assert actual_row["qt_version"] == expect_row["qt_version"] has_difference |= compare_modules_entry(actual_row, expect_row) continue actual_set = to_set(actual_combos[root_key]) expected_set = to_set(expected_combos[root_key]) has_difference |= report_difference( expected_set, actual_set, actual_name, root_key ) has_difference |= report_difference( actual_set, expected_set, expect_name, root_key ) return has_difference def alphabetize_modules(combos: Dict[str, Union[List[Dict], List[str]]]): for i, item in enumerate(combos["modules"]): combos["modules"][i]["modules"] = sorted(item["modules"]) def write_combinations_json( combos: Dict[str, Union[List[Dict], List[str]]], filename: Path, is_use_pretty_print: bool = True, ): logger.info(f"Write file {filename}") json_text = ( pretty_print_combos(combos) if is_use_pretty_print else json.dumps([combos], sort_keys=True, indent=2) ) if filename.write_text(json_text, encoding="utf_8") == 0: raise RuntimeError("Failed to write file!") def main(filename: Path, is_write_file: bool) -> int: try: expect = json.loads(filename.read_text()) alphabetize_modules(expect[0]) actual = generate_combos(new_archive=expect[0]["new_archive"]) logger.info("=" * 80) logger.info("Program Output:") logger.info(pretty_print_combos(actual)) logger.info("=" * 80) logger.info(f"Comparison with existing '{filename}':") diff = compare_combos(actual, expect[0], "program_output", str(filename)) logger.info("=" * 80) if not diff: print(f"{filename} is up to date! No PR is necessary this time!") return 0 # no difference if is_write_file: print(f"{filename} has changed; writing changes to file...") write_combinations_json(actual, filename) return 0 # file written successfully return 1 # difference reported except (ArchiveConnectionError, ArchiveDownloadError) as e: logger.error(format(e)) return 1 def get_tqdm(disable: bool): if disable: return lambda x: x from tqdm import tqdm as base_tqdm return lambda *a: base_tqdm(*a, disable=disable) if __name__ == "__main__": Settings.load_settings() setup_logging() logger = logging.getLogger("aqt.generate_combos") json_filename = Path(__file__).parent.parent / "aqt/combinations.json" parser = argparse.ArgumentParser( description="Generate combinations.json from download.qt.io, " "compare with existing file, and write file to correct differences" ) parser.add_argument( "--write", help="write to combinations.json if changes detected", action="store_true", ) parser.add_argument( "--no-tqdm", help="disable progress bars (makes CI logs easier to read)", action="store_true", ) args = parser.parse_args() tqdm = get_tqdm(args.no_tqdm) exit(main(filename=json_filename, is_write_file=args.write))