files.new_contents: #!/usr/bin/env python
  #
  # Copyright (C) 2018 Linus Jahn <lnj@kaidan.im>
  # Copyright (C) 2019-2021 Hiroshi Miura <miurahr@linux.com>
  #
  # Permission is hereby granted, free of charge, to any person obtaining a copy of
  # this software and associated documentation files (the "Software"), to deal in
  # the Software without restriction, including without limitation the rights to
  # use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
  # the Software, and to permit persons to whom the Software is furnished to do so,
  # subject to the following conditions:
  #
  # The above copyright notice and this permission notice shall be included in all
  # copies or substantial portions of the Software.
  #
  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
  # FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
  # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
  # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  import posixpath
  from dataclasses import dataclass, field
  from logging import getLogger
  from typing import Dict, Iterable, List, Optional, Tuple
  
  from defusedxml import ElementTree
  
  from aqt.exceptions import ArchiveDownloadError, ArchiveListError, NoPackageFound
  from aqt.helper import Settings, get_hash, getUrl, ssplit
  from aqt.metadata import QtRepoProperty, Version
  
  
  @dataclass
  class TargetConfig:
      version: str
      target: str
      arch: str
      os_name: str
  
      def __post_init__(self):
          self.version = str(self.version)
  
  
  @dataclass
  class QtPackage:
      name: str
      base_url: str
      archive_path: str
      archive: str
      package_desc: str
      pkg_update_name: str
      version: Optional[Version] = field(default=None)
  
      def __repr__(self):
          v_info = f", version={self.version}" if self.version else ""
          return f"QtPackage(name={self.name}, archive={self.archive}{v_info})"
  
      def __str__(self):
          v_info = f", version={self.version}" if self.version else ""
          return (
              f"QtPackage(name={self.name}, url={self.archive_path}, "
              f"archive={self.archive}, desc={self.package_desc}"
              f"{v_info})"
          )
  
  
  class ModuleToPackage:
      """
      Holds a mapping of module names to a list of Updates.xml PackageUpdate names.
      For example, we could have the following:
      {"qtcharts": ["qt.qt6.620.addons.qtcharts.arch", qt.qt6.620.qtcharts.arch", qt.620.addons.qtcharts.arch",])
      It also contains a reverse mapping of PackageUpdate names to module names, so that
      lookup of a package name and removal of a module name can be done in constant time.
      Without this reverse mapping, QtArchives._parse_update_xml would run at least one
      linear search on the forward mapping for each module installed.
  
      The list of PackageUpdate names consists of all the possible names for the PackageUpdate.
      The naming conventions for each PackageUpdate are not predictable, so we need to maintain
      a list of possibilities. While reading Updates.xml, if we encounter any one of the package
      names on this list, we can use it to install the package "qtcharts".
  
      Once we have installed the package, we need to remove the package "qtcharts" from this
      mapping, so we can keep track of what still needs to be installed.
      """
  
      def __init__(self, initial_map: Dict[str, List[str]]):
          self._modules_to_packages: Dict[str, List[str]] = initial_map
          self._packages_to_modules: Dict[str, str] = {
              value: key for key, list_of_values in initial_map.items() for value in list_of_values
          }
  
      def add(self, module_name: str, package_names: List[str]):
          self._modules_to_packages[module_name] = self._modules_to_packages.get(module_name, []) + package_names
          for package_name in package_names:
              assert package_name not in self._packages_to_modules, "Detected a package name collision"
              self._packages_to_modules[package_name] = module_name
  
      def remove_module_for_package(self, package_name: str):
          module_name = self._packages_to_modules[package_name]
          for package_name in self._modules_to_packages[module_name]:
              self._packages_to_modules.pop(package_name)
          self._modules_to_packages.pop(module_name)
  
      def has_package(self, package_name: str):
          return package_name in self._packages_to_modules
  
      def get_modules(self) -> Iterable[str]:
          return self._modules_to_packages.keys()
  
      def __len__(self) -> int:
          return len(self._modules_to_packages)
  
      def __format__(self, format_spec) -> str:
          return str(sorted(set(self._modules_to_packages.keys())))
  
  
  class QtArchives:
      """Download and hold Qt archive packages list.
      It access to download.qt.io site and get Update.xml file.
      It parse XML file and store metadata into list of QtPackage object.
      """
  
      def __init__(
          self,
          os_name: str,
          target: str,
          version_str: str,
          arch: str,
          base: str,
          subarchives: Optional[Iterable[str]] = None,
          modules: Optional[Iterable[str]] = None,
          all_extra: bool = False,
          is_include_base_package: bool = True,
          timeout=(5, 5),
      ):
          self.version: Version = Version(version_str)
          self.target: str = target
          self.arch: str = arch
          self.os_name: str = os_name
          self.all_extra: bool = all_extra
          self.arch_list: List[str] = [item.get("arch") for item in Settings.qt_combinations]
          self.base: str = base
          self.logger = getLogger("aqt.archives")
          self.archives: List[QtPackage] = []
          self.subarchives: Optional[Iterable[str]] = subarchives
          self.mod_list: Iterable[str] = modules or []
          self.is_include_base_package: bool = is_include_base_package
          self.timeout = timeout
          try:
              self._get_archives()
          except ArchiveDownloadError as e:
              self.handle_missing_updates_xml(e)
  
      def handle_missing_updates_xml(self, e: ArchiveDownloadError):
          msg = f"Failed to locate XML data for Qt version '{self.version}'."
          help_msg = f"Please use 'aqt list-qt {self.os_name} {self.target}' to show versions available."
          raise ArchiveListError(msg, suggested_action=[help_msg]) from e
  
      def should_filter_archives(self, package_name: str) -> bool:
          """
          This tells us, based on the PackageUpdate.Name property, whether or not the `self.subarchives`
          list should be used to filter out archives that we are not interested in.
  
          If `package_name` is a base module or a debug_info module, the `subarchives` list will apply to it.
          """
          return package_name in self._base_package_names() or "debug_info" in package_name
  
      def _version_str(self) -> str:
          return ("{0.major}{0.minor}" if self.version == Version("5.9.0") else "{0.major}{0.minor}{0.patch}").format(
              self.version
          )
  
      def _arch_ext(self) -> str:
          ext = QtRepoProperty.extension_for_arch(self.arch, self.version >= Version("6.0.0"))
          return ("_" + ext) if ext else ""
  
      def _base_module_name(self) -> str:
          """
          This is the name for the base Qt module, whose PackageUpdate.Name property would be
          'qt.123.gcc_64' or 'qt.qt1.123.gcc_64' for Qt 1.2.3, for architecture gcc_64.
          """
          return "qt_base"
  
      def _base_package_names(self) -> Iterable[str]:
          """
          This is a list of all potential PackageUpdate.Name properties for the base Qt module,
          which would be 'qt.123.gcc_64' or 'qt.qt1.123.gcc_64' for Qt 1.2.3, for architecture gcc_64,
          or 'qt.123.src' or 'qt.qt1.123.src' for the source module.
          """
          return (
              f"qt.qt{self.version.major}.{self._version_str()}.{self.arch}",
              f"qt.{self._version_str()}.{self.arch}",
          )
  
      def _module_name_suffix(self, module: str) -> str:
          return f"{module}.{self.arch}"
  
      def _target_packages(self) -> ModuleToPackage:
          if self.all_extra:
              return ModuleToPackage({})
          base_package = {self._base_module_name(): list(self._base_package_names())}
          target_packages = ModuleToPackage(base_package if self.is_include_base_package else {})
          if self.all_extra:
              return target_packages
          for module in self.mod_list:
              suffix = self._module_name_suffix(module)
              package_names = [
                  f"qt.qt{self.version.major}.{self._version_str()}.{suffix}",
                  f"qt.{self._version_str()}.{suffix}",
              ]
              if not module.startswith("addons."):
                  package_names.append(f"qt.qt{self.version.major}.{self._version_str()}.addons.{suffix}")
              target_packages.add(module, package_names)
          return target_packages
  
      def _get_archives(self):
          # Get packages index
  
          # os_target_folder: online/qtsdkrepository/windows_x86/desktop/qt5_59_src_doc_examples
          os_target_folder = posixpath.join(
              "online/qtsdkrepository",
              self.os_name + ("_x86" if self.os_name == "windows" else "_x64"),
              self.target,
              f"qt{self.version.major}_{self._version_str()}{self._arch_ext()}",
          )
          update_xml_path = posixpath.join(os_target_folder, "Updates.xml")
          self._download_update_xml(update_xml_path)
          self._parse_update_xml(os_target_folder, self._target_packages())
  
      def _download_update_xml(self, update_xml_path):
          """Hook for unit test."""
          xml_hash = get_hash(update_xml_path, "sha256", self.timeout)
          update_xml_text = getUrl(posixpath.join(self.base, update_xml_path), self.timeout, xml_hash)
          self.update_xml_text = update_xml_text
  
      def _parse_update_xml(self, os_target_folder, target_packages: Optional[ModuleToPackage]):
          if not target_packages:
              target_packages = ModuleToPackage({})
          try:
              self.update_xml = ElementTree.fromstring(self.update_xml_text)
          except ElementTree.ParseError as perror:
              raise ArchiveListError(f"Downloaded metadata is corrupted. {perror}") from perror
  
          for packageupdate in self.update_xml.iter("PackageUpdate"):
              pkg_name = packageupdate.find("Name").text
              downloads_text = packageupdate.find("DownloadableArchives").text
              if not downloads_text:
                  continue
              # If we asked for `--noarchives`, we don't want the base module
              if not self.is_include_base_package and pkg_name in self._base_package_names():
                  continue
              # Need to filter archives to download when we want all extra modules
              if self.all_extra:
                  # Check platform
                  name_last_section = pkg_name.split(".")[-1]
                  if name_last_section in self.arch_list and self.arch != name_last_section:
                      continue
                  # Check doc/examples
                  if self.arch in ["doc", "examples"]:
                      if self.arch not in pkg_name:
                          continue
              elif not target_packages.has_package(pkg_name):
                  continue
              else:
                  target_packages.remove_module_for_package(pkg_name)
              full_version = packageupdate.find("Version").text
              package_desc = packageupdate.find("Description").text
              should_filter_archives: bool = self.subarchives and self.should_filter_archives(pkg_name)
  
              for archive in ssplit(downloads_text):
                  archive_name = archive.split("-", maxsplit=1)[0]
                  if should_filter_archives and archive_name not in self.subarchives:
                      continue
                  archive_path = posixpath.join(
                      # online/qtsdkrepository/linux_x64/desktop/qt5_5150/
                      os_target_folder,
                      # qt.qt5.5150.gcc_64/
                      pkg_name,
                      # 5.15.0-0-202005140804qtbase-Linux-RHEL_7_6-GCC-Linux-RHEL_7_6-X86_64.7z
                      full_version + archive,
                  )
                  self.archives.append(
                      QtPackage(
                          name=archive_name,
                          base_url=self.base,
                          archive_path=archive_path,
                          archive=archive,
                          package_desc=package_desc,
                          pkg_update_name=pkg_name,  # For testing purposes
                      )
                  )
  
          # if we have located every requested package, then target_packages will be empty
          if len(target_packages) > 0:
              message = f"The packages {target_packages} were not found while parsing XML of package information!"
              raise NoPackageFound(message, suggested_action=self.help_msg(target_packages.get_modules()))
  
      def help_msg(self, missing_modules: Iterable[str]) -> Iterable[str]:
          base_cmd = f"aqt list-qt {self.os_name} {self.target}"
          arch = f"Please use '{base_cmd} --arch {self.version}' to show architectures available."
          mods = f"Please use '{base_cmd} --modules {self.version} <arch>' to show modules available."
          has_base_pkg: bool = self._base_module_name() in missing_modules
          has_non_base_pkg: bool = len(list(missing_modules)) > 1 or not has_base_pkg
          messages = []
          if has_base_pkg:
              messages.append(arch)
          if has_non_base_pkg:
              messages.append(mods)
          return messages
  
      def get_packages(self) -> List[QtPackage]:
          """
           It returns an archive package list.
  
          :return package list
          :rtype: List[QtPackage]
          """
          return self.archives
  
      def get_target_config(self) -> TargetConfig:
          """Get target configuration
  
          :return: configured target and its version with arch
          :rtype: TargetConfig object
          """
          return TargetConfig(self.version, self.target, self.arch, self.os_name)
  
  
  class SrcDocExamplesArchives(QtArchives):
      """Hold doc/src/example archive package list."""
  
      def __init__(
          self,
          flavor,
          os_name,
          target,
          version,
          base,
          subarchives=None,
          modules=None,
          all_extra=False,
          is_include_base_package: bool = True,
          timeout=(5, 5),
      ):
          self.flavor = flavor
          self.target = target
          self.os_name = os_name
          self.base = base
          self.logger = getLogger("aqt.archives")
          super(SrcDocExamplesArchives, self).__init__(
              os_name,
              target,
              version,
              arch=self.flavor,
              base=base,
              subarchives=subarchives,
              modules=modules,
              all_extra=all_extra,
              is_include_base_package=is_include_base_package,
              timeout=timeout,
          )
  
      def _arch_ext(self) -> str:
          return "_src_doc_examples"
  
      def _base_module_name(self) -> str:
          """
          This is the name for the base Qt Src/Doc/Example module, whose PackageUpdate.Name
          property would be 'qt.123.examples' or 'qt.qt1.123.examples' for Qt 1.2.3 examples.
          """
          return self.flavor  # src | doc | examples
  
      def _module_name_suffix(self, module: str) -> str:
          return f"{self.flavor}.{module}"
  
      def get_target_config(self) -> TargetConfig:
          """Get target configuration.
  
          :return tuple of three parameter, "src_doc_examples", target and arch
          """
          return TargetConfig("src_doc_examples", self.target, self.arch, self.os_name)
  
      def help_msg(self, missing_modules: Iterable[str]) -> Iterable[str]:
          cmd_type = "example" if self.flavor == "examples" else self.flavor
          base_cmd = f"aqt list-{cmd_type} {self.os_name} {self.version}"
          mods = f"Please use '{base_cmd} --modules' to show modules available."
          has_non_base_pkg: bool = len(list(missing_modules)) > 1
          messages = []
          if has_non_base_pkg:
              messages.append(mods)
          return messages
  
  
  class ToolArchives(QtArchives):
      """Hold tool archive package list
      when installing mingw tool, argument would be
      ToolArchive(windows, desktop, 4.9.1-3, mingw)
      when installing ifw tool, argument would be
      ToolArchive(linux, desktop, 3.1.1, ifw)
      """
  
      def __init__(
          self,
          os_name: str,
          target: str,
          tool_name: str,
          base: str,
          version_str: Optional[str] = None,
          arch: Optional[str] = None,
          timeout: Tuple[int, int] = (5, 5),
      ):
          self.tool_name = tool_name
          self.os_name = os_name
          self.logger = getLogger("aqt.archives")
          self.tool_version_str: Optional[str] = version_str
          super(ToolArchives, self).__init__(
              os_name=os_name,
              target=target,
              version_str="0.0.1",  # dummy value
              arch=arch,
              base=base,
              timeout=timeout,
          )
  
      def __str__(self):
          return f"ToolArchives(tool_name={self.tool_name}, version={self.version}, arch={self.arch})"
  
      def handle_missing_updates_xml(self, e: ArchiveDownloadError):
          msg = f"Failed to locate XML data for the tool '{self.tool_name}'."
          help_msg = f"Please use 'aqt list-tool {self.os_name} {self.target}' to show tools available."
          raise ArchiveListError(msg, suggested_action=[help_msg]) from e
  
      def _get_archives(self):
          os_target_folder = posixpath.join(
              "online/qtsdkrepository",
              # linux_x64/
              self.os_name + ("_x86" if self.os_name == "windows" else "_x64"),
              # desktop/
              self.target,
              # tools_ifw/
              self.tool_name,
          )
          update_xml_url = posixpath.join(os_target_folder, "Updates.xml")
          self._download_update_xml(update_xml_url)  # call super method.
          self._parse_update_xml(os_target_folder, None)
  
      def _parse_update_xml(self, os_target_folder, *ignored):
          try:
              self.update_xml = ElementTree.fromstring(self.update_xml_text)
          except ElementTree.ParseError as perror:
              raise ArchiveListError(f"Downloaded metadata is corrupted. {perror}") from perror
  
          try:
              packageupdate = next(filter(lambda x: x.find("Name").text == self.arch, self.update_xml.iter("PackageUpdate")))
          except StopIteration:
              message = f"The package '{self.arch}' was not found while parsing XML of package information!"
              raise NoPackageFound(message, suggested_action=self.help_msg())
  
          name = packageupdate.find("Name").text
          named_version = packageupdate.find("Version").text
          if self.tool_version_str and named_version != self.tool_version_str:
              message = f"The package '{self.arch}' has the version '{named_version}', not the requested '{self.version}'."
              raise NoPackageFound(message, suggested_action=self.help_msg())
          package_desc = packageupdate.find("Description").text
          downloadable_archives = packageupdate.find("DownloadableArchives").text
          if not downloadable_archives:
              message = f"The package '{self.arch}' contains no downloadable archives!"
              raise NoPackageFound(message)
          for archive in ssplit(downloadable_archives):
              archive_path = posixpath.join(
                  # online/qtsdkrepository/linux_x64/desktop/tools_ifw/
                  os_target_folder,
                  # qt.tools.ifw.41/
                  name,
                  #  4.1.1-202105261130ifw-linux-x64.7z
                  f"{named_version}{archive}",
              )
              self.archives.append(
                  QtPackage(
                      name=name,
                      base_url=self.base,
                      archive_path=archive_path,
                      archive=archive,
                      package_desc=package_desc,
                      pkg_update_name=name,  # Redundant
                  )
              )
  
      def help_msg(self, *args) -> Iterable[str]:
          return [f"Please use 'aqt list-tool {self.os_name} {self.target} {self.tool_name}' to show tool variants available."]
  
      def get_target_config(self) -> TargetConfig:
          """Get target configuration.
  
          :return tuple of three parameter, "Tools", target and arch
          """
          return TargetConfig("Tools", self.target, self.arch, self.os_name)
  
files.new_path: aqt/archives.py
files.old_contents: #!/usr/bin/env python
  #
  # Copyright (C) 2018 Linus Jahn <lnj@kaidan.im>
  # Copyright (C) 2019-2021 Hiroshi Miura <miurahr@linux.com>
  #
  # Permission is hereby granted, free of charge, to any person obtaining a copy of
  # this software and associated documentation files (the "Software"), to deal in
  # the Software without restriction, including without limitation the rights to
  # use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
  # the Software, and to permit persons to whom the Software is furnished to do so,
  # subject to the following conditions:
  #
  # The above copyright notice and this permission notice shall be included in all
  # copies or substantial portions of the Software.
  #
  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
  # FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
  # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
  # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  import binascii
  import posixpath
  from dataclasses import dataclass, field
  from logging import getLogger
  from typing import Dict, Iterable, List, Optional, Tuple
  
  from defusedxml import ElementTree
  
  from aqt.exceptions import ArchiveDownloadError, ArchiveListError, ChecksumDownloadFailure, NoPackageFound
  from aqt.helper import Settings, get_hash, getUrl, ssplit
  from aqt.metadata import QtRepoProperty, Version
  
  
  @dataclass
  class TargetConfig:
      version: str
      target: str
      arch: str
      os_name: str
  
      def __post_init__(self):
          self.version = str(self.version)
  
  
  @dataclass
  class QtPackage:
      name: str
      base_url: str
      archive_path: str
      archive: str
      package_desc: str
      pkg_update_name: str
      version: Optional[Version] = field(default=None)
  
      def __repr__(self):
          v_info = f", version={self.version}" if self.version else ""
          return f"QtPackage(name={self.name}, archive={self.archive}{v_info})"
  
      def __str__(self):
          v_info = f", version={self.version}" if self.version else ""
          return (
              f"QtPackage(name={self.name}, url={self.archive_path}, "
              f"archive={self.archive}, desc={self.package_desc}"
              f"{v_info})"
          )
  
  
  class ModuleToPackage:
      """
      Holds a mapping of module names to a list of Updates.xml PackageUpdate names.
      For example, we could have the following:
      {"qtcharts": ["qt.qt6.620.addons.qtcharts.arch", qt.qt6.620.qtcharts.arch", qt.620.addons.qtcharts.arch",])
      It also contains a reverse mapping of PackageUpdate names to module names, so that
      lookup of a package name and removal of a module name can be done in constant time.
      Without this reverse mapping, QtArchives._parse_update_xml would run at least one
      linear search on the forward mapping for each module installed.
  
      The list of PackageUpdate names consists of all the possible names for the PackageUpdate.
      The naming conventions for each PackageUpdate are not predictable, so we need to maintain
      a list of possibilities. While reading Updates.xml, if we encounter any one of the package
      names on this list, we can use it to install the package "qtcharts".
  
      Once we have installed the package, we need to remove the package "qtcharts" from this
      mapping, so we can keep track of what still needs to be installed.
      """
  
      def __init__(self, initial_map: Dict[str, List[str]]):
          self._modules_to_packages: Dict[str, List[str]] = initial_map
          self._packages_to_modules: Dict[str, str] = {
              value: key for key, list_of_values in initial_map.items() for value in list_of_values
          }
  
      def add(self, module_name: str, package_names: List[str]):
          self._modules_to_packages[module_name] = self._modules_to_packages.get(module_name, []) + package_names
          for package_name in package_names:
              assert package_name not in self._packages_to_modules, "Detected a package name collision"
              self._packages_to_modules[package_name] = module_name
  
      def remove_module_for_package(self, package_name: str):
          module_name = self._packages_to_modules[package_name]
          for package_name in self._modules_to_packages[module_name]:
              self._packages_to_modules.pop(package_name)
          self._modules_to_packages.pop(module_name)
  
      def has_package(self, package_name: str):
          return package_name in self._packages_to_modules
  
      def get_modules(self) -> Iterable[str]:
          return self._modules_to_packages.keys()
  
      def __len__(self) -> int:
          return len(self._modules_to_packages)
  
      def __format__(self, format_spec) -> str:
          return str(sorted(set(self._modules_to_packages.keys())))
  
  
  class QtArchives:
      """Download and hold Qt archive packages list.
      It access to download.qt.io site and get Update.xml file.
      It parse XML file and store metadata into list of QtPackage object.
      """
  
      def __init__(
          self,
          os_name: str,
          target: str,
          version_str: str,
          arch: str,
          base: str,
          subarchives: Optional[Iterable[str]] = None,
          modules: Optional[Iterable[str]] = None,
          all_extra: bool = False,
          is_include_base_package: bool = True,
          timeout=(5, 5),
      ):
          self.version: Version = Version(version_str)
          self.target: str = target
          self.arch: str = arch
          self.os_name: str = os_name
          self.all_extra: bool = all_extra
          self.arch_list: List[str] = [item.get("arch") for item in Settings.qt_combinations]
          self.base: str = base
          self.logger = getLogger("aqt.archives")
          self.archives: List[QtPackage] = []
          self.subarchives: Optional[Iterable[str]] = subarchives
          self.mod_list: Iterable[str] = modules or []
          self.is_include_base_package: bool = is_include_base_package
          self.timeout = timeout
          try:
              self._get_archives()
          except ArchiveDownloadError as e:
              self.handle_missing_updates_xml(e)
  
      def handle_missing_updates_xml(self, e: ArchiveDownloadError):
          msg = f"Failed to locate XML data for Qt version '{self.version}'."
          help_msg = f"Please use 'aqt list-qt {self.os_name} {self.target}' to show versions available."
          raise ArchiveListError(msg, suggested_action=[help_msg]) from e
  
      def should_filter_archives(self, package_name: str) -> bool:
          """
          This tells us, based on the PackageUpdate.Name property, whether or not the `self.subarchives`
          list should be used to filter out archives that we are not interested in.
  
          If `package_name` is a base module or a debug_info module, the `subarchives` list will apply to it.
          """
          return package_name in self._base_package_names() or "debug_info" in package_name
  
      def _version_str(self) -> str:
          return ("{0.major}{0.minor}" if self.version == Version("5.9.0") else "{0.major}{0.minor}{0.patch}").format(
              self.version
          )
  
      def _arch_ext(self) -> str:
          ext = QtRepoProperty.extension_for_arch(self.arch, self.version >= Version("6.0.0"))
          return ("_" + ext) if ext else ""
  
      def _base_module_name(self) -> str:
          """
          This is the name for the base Qt module, whose PackageUpdate.Name property would be
          'qt.123.gcc_64' or 'qt.qt1.123.gcc_64' for Qt 1.2.3, for architecture gcc_64.
          """
          return "qt_base"
  
      def _base_package_names(self) -> Iterable[str]:
          """
          This is a list of all potential PackageUpdate.Name properties for the base Qt module,
          which would be 'qt.123.gcc_64' or 'qt.qt1.123.gcc_64' for Qt 1.2.3, for architecture gcc_64,
          or 'qt.123.src' or 'qt.qt1.123.src' for the source module.
          """
          return (
              f"qt.qt{self.version.major}.{self._version_str()}.{self.arch}",
              f"qt.{self._version_str()}.{self.arch}",
          )
  
      def _module_name_suffix(self, module: str) -> str:
          return f"{module}.{self.arch}"
  
      def _target_packages(self) -> ModuleToPackage:
          if self.all_extra:
              return ModuleToPackage({})
          base_package = {self._base_module_name(): list(self._base_package_names())}
          target_packages = ModuleToPackage(base_package if self.is_include_base_package else {})
          if self.all_extra:
              return target_packages
          for module in self.mod_list:
              suffix = self._module_name_suffix(module)
              package_names = [
                  f"qt.qt{self.version.major}.{self._version_str()}.{suffix}",
                  f"qt.{self._version_str()}.{suffix}",
              ]
              if not module.startswith("addons."):
                  package_names.append(f"qt.qt{self.version.major}.{self._version_str()}.addons.{suffix}")
              target_packages.add(module, package_names)
          return target_packages
  
      def _get_archives(self):
          # Get packages index
  
          # os_target_folder: online/qtsdkrepository/windows_x86/desktop/qt5_59_src_doc_examples
          os_target_folder = posixpath.join(
              "online/qtsdkrepository",
              self.os_name + ("_x86" if self.os_name == "windows" else "_x64"),
              self.target,
              f"qt{self.version.major}_{self._version_str()}{self._arch_ext()}",
          )
          update_xml_path = posixpath.join(os_target_folder, "Updates.xml")
          self._download_update_xml(update_xml_path)
          self._parse_update_xml(os_target_folder, self._target_packages())
  
      def _download_update_xml(self, update_xml_path):
          """Hook for unit test."""
          xml_hash = binascii.unhexlify(get_hash(update_xml_path, "sha256", self.timeout))
          if xml_hash == "":
              raise ChecksumDownloadFailure(f"Checksum for '{update_xml_path}' is empty")
          update_xml_text = getUrl(posixpath.join(self.base, update_xml_path), self.timeout, xml_hash)
          self.update_xml_text = update_xml_text
  
      def _parse_update_xml(self, os_target_folder, target_packages: Optional[ModuleToPackage]):
          if not target_packages:
              target_packages = ModuleToPackage({})
          try:
              self.update_xml = ElementTree.fromstring(self.update_xml_text)
          except ElementTree.ParseError as perror:
              raise ArchiveListError(f"Downloaded metadata is corrupted. {perror}") from perror
  
          for packageupdate in self.update_xml.iter("PackageUpdate"):
              pkg_name = packageupdate.find("Name").text
              downloads_text = packageupdate.find("DownloadableArchives").text
              if not downloads_text:
                  continue
              # If we asked for `--noarchives`, we don't want the base module
              if not self.is_include_base_package and pkg_name in self._base_package_names():
                  continue
              # Need to filter archives to download when we want all extra modules
              if self.all_extra:
                  # Check platform
                  name_last_section = pkg_name.split(".")[-1]
                  if name_last_section in self.arch_list and self.arch != name_last_section:
                      continue
                  # Check doc/examples
                  if self.arch in ["doc", "examples"]:
                      if self.arch not in pkg_name:
                          continue
              elif not target_packages.has_package(pkg_name):
                  continue
              else:
                  target_packages.remove_module_for_package(pkg_name)
              full_version = packageupdate.find("Version").text
              package_desc = packageupdate.find("Description").text
              should_filter_archives: bool = self.subarchives and self.should_filter_archives(pkg_name)
  
              for archive in ssplit(downloads_text):
                  archive_name = archive.split("-", maxsplit=1)[0]
                  if should_filter_archives and archive_name not in self.subarchives:
                      continue
                  archive_path = posixpath.join(
                      # online/qtsdkrepository/linux_x64/desktop/qt5_5150/
                      os_target_folder,
                      # qt.qt5.5150.gcc_64/
                      pkg_name,
                      # 5.15.0-0-202005140804qtbase-Linux-RHEL_7_6-GCC-Linux-RHEL_7_6-X86_64.7z
                      full_version + archive,
                  )
                  self.archives.append(
                      QtPackage(
                          name=archive_name,
                          base_url=self.base,
                          archive_path=archive_path,
                          archive=archive,
                          package_desc=package_desc,
                          pkg_update_name=pkg_name,  # For testing purposes
                      )
                  )
  
          # if we have located every requested package, then target_packages will be empty
          if len(target_packages) > 0:
              message = f"The packages {target_packages} were not found while parsing XML of package information!"
              raise NoPackageFound(message, suggested_action=self.help_msg(target_packages.get_modules()))
  
      def help_msg(self, missing_modules: Iterable[str]) -> Iterable[str]:
          base_cmd = f"aqt list-qt {self.os_name} {self.target}"
          arch = f"Please use '{base_cmd} --arch {self.version}' to show architectures available."
          mods = f"Please use '{base_cmd} --modules {self.version} <arch>' to show modules available."
          has_base_pkg: bool = self._base_module_name() in missing_modules
          has_non_base_pkg: bool = len(list(missing_modules)) > 1 or not has_base_pkg
          messages = []
          if has_base_pkg:
              messages.append(arch)
          if has_non_base_pkg:
              messages.append(mods)
          return messages
  
      def get_packages(self) -> List[QtPackage]:
          """
           It returns an archive package list.
  
          :return package list
          :rtype: List[QtPackage]
          """
          return self.archives
  
      def get_target_config(self) -> TargetConfig:
          """Get target configuration
  
          :return: configured target and its version with arch
          :rtype: TargetConfig object
          """
          return TargetConfig(self.version, self.target, self.arch, self.os_name)
  
  
  class SrcDocExamplesArchives(QtArchives):
      """Hold doc/src/example archive package list."""
  
      def __init__(
          self,
          flavor,
          os_name,
          target,
          version,
          base,
          subarchives=None,
          modules=None,
          all_extra=False,
          is_include_base_package: bool = True,
          timeout=(5, 5),
      ):
          self.flavor = flavor
          self.target = target
          self.os_name = os_name
          self.base = base
          self.logger = getLogger("aqt.archives")
          super(SrcDocExamplesArchives, self).__init__(
              os_name,
              target,
              version,
              arch=self.flavor,
              base=base,
              subarchives=subarchives,
              modules=modules,
              all_extra=all_extra,
              is_include_base_package=is_include_base_package,
              timeout=timeout,
          )
  
      def _arch_ext(self) -> str:
          return "_src_doc_examples"
  
      def _base_module_name(self) -> str:
          """
          This is the name for the base Qt Src/Doc/Example module, whose PackageUpdate.Name
          property would be 'qt.123.examples' or 'qt.qt1.123.examples' for Qt 1.2.3 examples.
          """
          return self.flavor  # src | doc | examples
  
      def _module_name_suffix(self, module: str) -> str:
          return f"{self.flavor}.{module}"
  
      def get_target_config(self) -> TargetConfig:
          """Get target configuration.
  
          :return tuple of three parameter, "src_doc_examples", target and arch
          """
          return TargetConfig("src_doc_examples", self.target, self.arch, self.os_name)
  
      def help_msg(self, missing_modules: Iterable[str]) -> Iterable[str]:
          cmd_type = "example" if self.flavor == "examples" else self.flavor
          base_cmd = f"aqt list-{cmd_type} {self.os_name} {self.version}"
          mods = f"Please use '{base_cmd} --modules' to show modules available."
          has_non_base_pkg: bool = len(list(missing_modules)) > 1
          messages = []
          if has_non_base_pkg:
              messages.append(mods)
          return messages
  
  
  class ToolArchives(QtArchives):
      """Hold tool archive package list
      when installing mingw tool, argument would be
      ToolArchive(windows, desktop, 4.9.1-3, mingw)
      when installing ifw tool, argument would be
      ToolArchive(linux, desktop, 3.1.1, ifw)
      """
  
      def __init__(
          self,
          os_name: str,
          target: str,
          tool_name: str,
          base: str,
          version_str: Optional[str] = None,
          arch: Optional[str] = None,
          timeout: Tuple[int, int] = (5, 5),
      ):
          self.tool_name = tool_name
          self.os_name = os_name
          self.logger = getLogger("aqt.archives")
          self.tool_version_str: Optional[str] = version_str
          super(ToolArchives, self).__init__(
              os_name=os_name,
              target=target,
              version_str="0.0.1",  # dummy value
              arch=arch,
              base=base,
              timeout=timeout,
          )
  
      def __str__(self):
          return f"ToolArchives(tool_name={self.tool_name}, version={self.version}, arch={self.arch})"
  
      def handle_missing_updates_xml(self, e: ArchiveDownloadError):
          msg = f"Failed to locate XML data for the tool '{self.tool_name}'."
          help_msg = f"Please use 'aqt list-tool {self.os_name} {self.target}' to show tools available."
          raise ArchiveListError(msg, suggested_action=[help_msg]) from e
  
      def _get_archives(self):
          os_target_folder = posixpath.join(
              "online/qtsdkrepository",
              # linux_x64/
              self.os_name + ("_x86" if self.os_name == "windows" else "_x64"),
              # desktop/
              self.target,
              # tools_ifw/
              self.tool_name,
          )
          update_xml_url = posixpath.join(os_target_folder, "Updates.xml")
          self._download_update_xml(update_xml_url)  # call super method.
          self._parse_update_xml(os_target_folder, None)
  
      def _parse_update_xml(self, os_target_folder, *ignored):
          try:
              self.update_xml = ElementTree.fromstring(self.update_xml_text)
          except ElementTree.ParseError as perror:
              raise ArchiveListError(f"Downloaded metadata is corrupted. {perror}") from perror
  
          try:
              packageupdate = next(filter(lambda x: x.find("Name").text == self.arch, self.update_xml.iter("PackageUpdate")))
          except StopIteration:
              message = f"The package '{self.arch}' was not found while parsing XML of package information!"
              raise NoPackageFound(message, suggested_action=self.help_msg())
  
          name = packageupdate.find("Name").text
          named_version = packageupdate.find("Version").text
          if self.tool_version_str and named_version != self.tool_version_str:
              message = f"The package '{self.arch}' has the version '{named_version}', not the requested '{self.version}'."
              raise NoPackageFound(message, suggested_action=self.help_msg())
          package_desc = packageupdate.find("Description").text
          downloadable_archives = packageupdate.find("DownloadableArchives").text
          if not downloadable_archives:
              message = f"The package '{self.arch}' contains no downloadable archives!"
              raise NoPackageFound(message)
          for archive in ssplit(downloadable_archives):
              archive_path = posixpath.join(
                  # online/qtsdkrepository/linux_x64/desktop/tools_ifw/
                  os_target_folder,
                  # qt.tools.ifw.41/
                  name,
                  #  4.1.1-202105261130ifw-linux-x64.7z
                  f"{named_version}{archive}",
              )
              self.archives.append(
                  QtPackage(
                      name=name,
                      base_url=self.base,
                      archive_path=archive_path,
                      archive=archive,
                      package_desc=package_desc,
                      pkg_update_name=name,  # Redundant
                  )
              )
  
      def help_msg(self, *args) -> Iterable[str]:
          return [f"Please use 'aqt list-tool {self.os_name} {self.target} {self.tool_name}' to show tool variants available."]
  
      def get_target_config(self) -> TargetConfig:
          """Get target configuration.
  
          :return tuple of three parameter, "Tools", target and arch
          """
          return TargetConfig("Tools", self.target, self.arch, self.os_name)
  
files.old_path: aqt/archives.py
files.new_contents: #!/usr/bin/env python
  #
  # Copyright (C) 2019-2021 Hiroshi Miura <miurahr@linux.com>
  #
  # Permission is hereby granted, free of charge, to any person obtaining a copy of
  # this software and associated documentation files (the "Software"), to deal in
  # the Software without restriction, including without limitation the rights to
  # use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
  # the Software, and to permit persons to whom the Software is furnished to do so,
  # subject to the following conditions:
  #
  # The above copyright notice and this permission notice shall be included in all
  # copies or substantial portions of the Software.
  #
  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
  # FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
  # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
  # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  import binascii
  import configparser
  import hashlib
  import json
  import logging.config
  import os
  import posixpath
  import secrets
  import sys
  from logging import getLogger
  from logging.handlers import QueueListener
  from pathlib import Path
  from typing import Callable, Dict, Generator, List, Optional, Tuple
  from urllib.parse import urlparse
  from xml.etree.ElementTree import Element
  
  import requests
  import requests.adapters
  from defusedxml import ElementTree
  
  from aqt.exceptions import (
      ArchiveChecksumError,
      ArchiveConnectionError,
      ArchiveDownloadError,
      ArchiveListError,
      ChecksumDownloadFailure,
  )
  
  
  def _get_meta(url: str):
      return requests.get(url + ".meta4")
  
  
  def _check_content_type(ct: str) -> bool:
      candidate = ["application/metalink4+xml", "text/plain"]
      return any(ct.startswith(t) for t in candidate)
  
  
  def getUrl(url: str, timeout, expected_hash: Optional[bytes] = None) -> str:
      """
      Gets a file from `url` via HTTP GET.
  
      No caller should call this function without providing an expected_hash, unless
      the caller is `get_hash`, which cannot know what the expected hash should be.
      """
      logger = getLogger("aqt.helper")
      with requests.sessions.Session() as session:
          retries = requests.adapters.Retry(
              total=Settings.max_retries_on_connection_error, backoff_factor=Settings.backoff_factor
          )
          adapter = requests.adapters.HTTPAdapter(max_retries=retries)
          session.mount("http://", adapter)
          session.mount("https://", adapter)
          try:
              r = session.get(url, allow_redirects=False, timeout=timeout)
              num_redirects = 0
              while 300 < r.status_code < 309 and num_redirects < 10:
                  num_redirects += 1
                  logger.debug("Asked to redirect({}) to: {}".format(r.status_code, r.headers["Location"]))
                  newurl = altlink(r.url, r.headers["Location"])
                  logger.info("Redirected: {}".format(urlparse(newurl).hostname))
                  r = session.get(newurl, stream=True, timeout=timeout)
          except (
              ConnectionResetError,
              requests.exceptions.ConnectionError,
              requests.exceptions.Timeout,
          ) as e:
              raise ArchiveConnectionError(f"Failure to connect to {url}: {type(e).__name__}") from e
          else:
              if r.status_code != 200:
                  msg = f"Failed to retrieve file at {url}\nServer response code: {r.status_code}, reason: {r.reason}"
                  raise ArchiveDownloadError(msg)
          result = r.text
          filename = url.split("/")[-1]
          actual_hash = hashlib.sha256(bytes(result, "utf-8")).digest()
          if expected_hash is not None and expected_hash != actual_hash:
              raise ArchiveChecksumError(
                  f"Downloaded file {filename} is corrupted! Detect checksum error.\n"
                  f"Expect {expected_hash.hex()}: {url}\n"
                  f"Actual {actual_hash.hex()}: {filename}"
              )
      return result
  
  
  def downloadBinaryFile(url: str, out: Path, hash_algo: str, exp: bytes, timeout):
      logger = getLogger("aqt.helper")
      filename = Path(url).name
      with requests.sessions.Session() as session:
          retries = requests.adapters.Retry(
              total=Settings.max_retries_on_connection_error, backoff_factor=Settings.backoff_factor
          )
          adapter = requests.adapters.HTTPAdapter(max_retries=retries)
          session.mount("http://", adapter)
          session.mount("https://", adapter)
          try:
              r = session.get(url, allow_redirects=False, stream=True, timeout=timeout)
              if 300 < r.status_code < 309:
                  logger.debug("Asked to redirect({}) to: {}".format(r.status_code, r.headers["Location"]))
                  newurl = altlink(r.url, r.headers["Location"])
                  logger.info("Redirected: {}".format(urlparse(newurl).hostname))
                  r = session.get(newurl, stream=True, timeout=timeout)
          except requests.exceptions.ConnectionError as e:
              raise ArchiveConnectionError(f"Connection error: {e.args}") from e
          except requests.exceptions.Timeout as e:
              raise ArchiveConnectionError(f"Connection timeout: {e.args}") from e
          else:
              hash = hashlib.new(hash_algo)
              try:
                  with open(out, "wb") as fd:
                      for chunk in r.iter_content(chunk_size=8196):
                          fd.write(chunk)
                          hash.update(chunk)
                      fd.flush()
              except Exception as e:
                  raise ArchiveDownloadError(f"Download of {filename} has error: {e}") from e
              if exp is not None and hash.digest() != exp:
                  raise ArchiveChecksumError(
                      f"Downloaded file {filename} is corrupted! Detect checksum error.\n"
                      f"Expect {exp.hex()}: {url}\n"
                      f"Actual {hash.digest().hex()}: {out.name}"
                  )
  
  
  def retry_on_errors(action: Callable[[], any], acceptable_errors: Tuple, num_retries: int, name: str):
      logger = getLogger("aqt.helper")
      for i in range(num_retries):
          try:
              retry_msg = f": attempt #{1 + i}" if i > 0 else ""
              logger.info(f"{name}{retry_msg}...")
              ret = action()
              if i > 0:
                  logger.info(f"Success on attempt #{1 + i}: {name}")
              return ret
          except acceptable_errors as e:
              if i < num_retries - 1:
                  continue  # just try again
              raise e from e
  
  
  def retry_on_bad_connection(function: Callable[[str], any], base_url: str):
      logger = getLogger("aqt.helper")
      fallback_url = secrets.choice(Settings.fallbacks)
      try:
          return function(base_url)
      except ArchiveConnectionError:
          logger.warning(f"Connection to '{base_url}' failed. Retrying with fallback '{fallback_url}'.")
          return function(fallback_url)
  
  
  def iter_list_reps(_list: List, num_reps: int) -> Generator:
      list_index = 0
      for i in range(num_reps):
          yield _list[list_index]
          list_index += 1
          if list_index >= len(_list):
              list_index = 0
  
  
  def get_hash(archive_path: str, algorithm: str, timeout) -> bytes:
      """
      Downloads a checksum and unhexlifies it to a `bytes` object, guaranteed to be the right length.
      Raises ChecksumDownloadFailure if the download failed, or if the checksum was un unexpected length.
  
      :param archive_path: The path to the file that we want to check, not the path to the checksum.
      :param algorithm:    sha256 is the only safe value to use here.
      :param timeout:      The timeout used by getUrl.
      :return:             A checksum in `bytes`
      """
      logger = getLogger("aqt.helper")
      hash_lengths = {"sha256": 64, "sha1": 40, "md5": 32}
      for base_url in iter_list_reps(Settings.trusted_mirrors, Settings.max_retries_to_retrieve_hash):
          url = posixpath.join(base_url, f"{archive_path}.{algorithm}")
          logger.debug(f"Attempt to download checksum at {url}")
          try:
              r = getUrl(url, timeout)
              # sha256 & md5 files are: "some_hash archive_filename"
              _hash = r.split(" ")[0]
              if len(_hash) == hash_lengths[algorithm]:
                  return binascii.unhexlify(_hash)
          except (ArchiveConnectionError, ArchiveDownloadError, binascii.Incomplete, binascii.Error):
              pass
      filename = archive_path.split("/")[-1]
      raise ChecksumDownloadFailure(
          f"Failed to download checksum for the file '{filename}' from mirrors '{Settings.trusted_mirrors}"
      )
  
  
  def altlink(url: str, alt: str):
      """
      Blacklisting redirected(alt) location based on Settings.blacklist configuration.
      When found black url, then try download a url + .meta4 that is a metalink version4
      xml file, parse it and retrieve best alternative url.
      """
      logger = getLogger("aqt.helper")
      if not any(alt.startswith(b) for b in Settings.blacklist):
          return alt
      try:
          m = _get_meta(url)
      except requests.exceptions.ConnectionError:
          logger.error("Got connection error. Fall back to recovery plan...")
          return alt
      else:
          # Expected response->'application/metalink4+xml; charset=utf-8'
          if not _check_content_type(m.headers["content-type"]):
              logger.error("Unexpected meta4 response;content-type: {}".format(m.headers["content-type"]))
              return alt
          try:
              mirror_xml = ElementTree.fromstring(m.text)
              meta_urls = {}
              for f in mirror_xml.iter("{urn:ietf:params:xml:ns:metalink}file"):
                  for u in f.iter("{urn:ietf:params:xml:ns:metalink}url"):
                      meta_urls[u.attrib["priority"]] = u.text
              mirrors = [meta_urls[i] for i in sorted(meta_urls.keys(), key=lambda x: int(x))]
          except Exception:
              exc_info = sys.exc_info()
              logger.error("Unexpected meta4 file; parse error: {}".format(exc_info[1]))
              return alt
          else:
              # Return first priority item which is not blacklist in mirrors list,
              # if not found then return alt in default
              return next(
                  filter(
                      lambda mirror: not any(mirror.startswith(b) for b in Settings.blacklist),
                      mirrors,
                  ),
                  alt,
              )
  
  
  class MyQueueListener(QueueListener):
      def __init__(self, queue):
          handlers = []
          super().__init__(queue, *handlers)
  
      def handle(self, record):
          """
          Handle a record from subprocess.
          Override logger name then handle at proper logger.
          """
          record = self.prepare(record)
          logger = getLogger("aqt.installer")
          record.name = "aqt.installer"
          logger.handle(record)
  
  
  def ssplit(data: str):
      for element in data.split(","):
          yield element.strip()
  
  
  def xml_to_modules(
      xml_text: str,
      predicate: Callable[[Element], bool],
  ) -> Dict[str, Dict[str, str]]:
      """Converts an XML document to a dict of `PackageUpdate` dicts, indexed by `Name` attribute.
      Only report elements that satisfy `predicate(element)`.
      Reports all keys available in the PackageUpdate tag as strings.
  
      :param xml_text: The entire contents of an xml file
      :param predicate: A function that decides which elements to keep or discard
      """
      try:
          parsed_xml = ElementTree.fromstring(xml_text)
      except ElementTree.ParseError as perror:
          raise ArchiveListError(f"Downloaded metadata is corrupted. {perror}") from perror
      packages = {}
      for packageupdate in parsed_xml.iter("PackageUpdate"):
          if predicate and not predicate(packageupdate):
              continue
          name = packageupdate.find("Name").text
          packages[name] = {}
          for child in packageupdate:
              packages[name][child.tag] = child.text
      return packages
  
  
  class MyConfigParser(configparser.ConfigParser):
      def getlist(self, section: str, option: str, fallback=[]) -> List[str]:
          value = self.get(section, option, fallback=None)
          if value is None:
              return fallback
          try:
              result = list(filter(None, (x.strip() for x in value.splitlines())))
          except Exception:
              result = fallback
          return result
  
      def getlistint(self, section: str, option: str, fallback=[]):
          try:
              result = [int(x) for x in self.getlist(section, option)]
          except Exception:
              result = fallback
          return result
  
  
  class SettingsClass:
      """
      Class to hold configuration and settings.
      Actual values are stored in 'settings.ini' file.
      It also holds a combinations database.
      """
  
      def __init__(self):
          self.config = MyConfigParser()
          self.configfile = os.path.join(os.path.dirname(__file__), "settings.ini")
          self.loggingconf = os.path.join(os.path.dirname(__file__), "logging.ini")
  
      def load_settings(self, file=None):
          with open(
              os.path.join(os.path.dirname(__file__), "combinations.json"),
              "r",
          ) as j:
              self._combinations = json.load(j)[0]
          if file is not None:
              if isinstance(file, str):
                  result = self.config.read(file)
                  if len(result) == 0:
                      raise IOError("Fails to load specified config file {}".format(file))
                  self.configfile = file
              else:
                  # passed through command line argparse.FileType("r")
                  self.config.read_file(file)
                  self.configfile = file
                  file.close()
          else:
              if isinstance(self.configfile, str):
                  with open(self.configfile, "r") as f:
                      self.config.read_file(f)
              else:
                  self.configfile.seek(0)
                  self.config.read_file(self.configfile)
  
      @property
      def qt_combinations(self):
          return self._combinations["qt"]
  
      @property
      def tools_combinations(self):
          return self._combinations["tools"]
  
      @property
      def available_versions(self):
          return self._combinations["versions"]
  
      @property
      def available_offline_installer_version(self):
          res = self._combinations["new_archive"]
          res.extend(self._combinations["versions"])
          return res
  
      def available_modules(self, qt_version):
          """Known module names
  
          :returns: dictionary of qt_version and module names
          :rtype: List[str]
          """
          modules = self._combinations["modules"]
          versions = qt_version.split(".")
          version = "{}.{}".format(versions[0], versions[1])
          result = None
          for record in modules:
              if record["qt_version"] == version:
                  result = record["modules"]
          return result
  
      @property
      def archive_download_location(self):
          return self.config.get("aqt", "archive_download_location", fallback=".")
  
      @property
      def always_keep_archives(self):
          return self.config.getboolean("aqt", "always_keep_archives", fallback=False)
  
      @property
      def concurrency(self):
          """concurrency configuration.
  
          :return: concurrency
          :rtype: int
          """
          return self.config.getint("aqt", "concurrency", fallback=4)
  
      @property
      def blacklist(self):
          """list of sites in a blacklist
  
          :returns: list of site URLs(scheme and host part)
          :rtype: List[str]
          """
          return self.config.getlist("mirrors", "blacklist", fallback=[])
  
      @property
      def baseurl(self):
          return self.config.get("aqt", "baseurl", fallback="https://download.qt.io")
  
      @property
      def connection_timeout(self):
          return self.config.getfloat("requests", "connection_timeout", fallback=3.5)
  
      @property
      def response_timeout(self):
          return self.config.getfloat("requests", "response_timeout", fallback=10)
  
      @property
      def max_retries(self):
          """Deprecated: please use `max_retries_on_connection_error` and `max_retries_on_checksum_error` instead!"""
          return self.config.getfloat("requests", "max_retries", fallback=5)
  
      @property
      def max_retries_on_connection_error(self):
          return self.config.getfloat("requests", "max_retries_on_connection_error", fallback=self.max_retries)
  
      @property
      def max_retries_on_checksum_error(self):
          return self.config.getint("requests", "max_retries_on_checksum_error", fallback=int(self.max_retries))
  
      @property
      def max_retries_to_retrieve_hash(self):
          return self.config.getint("requests", "max_retries_to_retrieve_hash", fallback=int(self.max_retries))
  
      @property
      def backoff_factor(self):
          return self.config.getfloat("requests", "retry_backoff", fallback=0.1)
  
      @property
      def trusted_mirrors(self):
          return self.config.getlist("mirrors", "trusted_mirrors", fallback=[self.baseurl])
  
      @property
      def fallbacks(self):
          return self.config.getlist("mirrors", "fallbacks", fallback=[])
  
      @property
      def zipcmd(self):
          return self.config.get("aqt", "7zcmd", fallback="7z")
  
      @property
      def kde_patches(self):
          return self.config.getlist("kde_patches", "patches", fallback=[])
  
      @property
      def print_stacktrace_on_error(self):
          return self.config.getboolean("aqt", "print_stacktrace_on_error", fallback=False)
  
      @property
      def min_module_size(self):
          """
          Some modules in the Qt repository contain only empty directories.
          We have found that these modules are no more than 40 bytes after decompression.
          This setting is used to filter out these empty modules in `list-*` output.
          """
          return self.config.getint("aqt", "min_module_size", fallback=41)
  
  
  Settings = SettingsClass()
  
  
  def setup_logging(env_key="LOG_CFG"):
      config = os.getenv(env_key, None)
      if config is not None and os.path.exists(config):
          logging.config.fileConfig(config)
      else:
          logging.config.fileConfig(Settings.loggingconf)
  
files.new_path: aqt/helper.py
files.old_contents: #!/usr/bin/env python
  #
  # Copyright (C) 2019-2021 Hiroshi Miura <miurahr@linux.com>
  #
  # Permission is hereby granted, free of charge, to any person obtaining a copy of
  # this software and associated documentation files (the "Software"), to deal in
  # the Software without restriction, including without limitation the rights to
  # use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
  # the Software, and to permit persons to whom the Software is furnished to do so,
  # subject to the following conditions:
  #
  # The above copyright notice and this permission notice shall be included in all
  # copies or substantial portions of the Software.
  #
  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
  # FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
  # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
  # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  
  import configparser
  import hashlib
  import json
  import logging.config
  import os
  import posixpath
  import secrets
  import sys
  from logging import getLogger
  from logging.handlers import QueueListener
  from pathlib import Path
  from typing import Callable, Dict, Generator, List, Optional, Tuple
  from urllib.parse import urlparse
  from xml.etree.ElementTree import Element
  
  import requests
  import requests.adapters
  from defusedxml import ElementTree
  
  from aqt.exceptions import (
      ArchiveChecksumError,
      ArchiveConnectionError,
      ArchiveDownloadError,
      ArchiveListError,
      ChecksumDownloadFailure,
  )
  
  
  def _get_meta(url: str):
      return requests.get(url + ".meta4")
  
  
  def _check_content_type(ct: str) -> bool:
      candidate = ["application/metalink4+xml", "text/plain"]
      return any(ct.startswith(t) for t in candidate)
  
  
  def getUrl(url: str, timeout, expected_hash: Optional[bytes] = None) -> str:
      """
      Gets a file from `url` via HTTP GET.
  
      No caller should call this function without providing an expected_hash, unless
      the caller is `get_hash`, which cannot know what the expected hash should be.
      """
      logger = getLogger("aqt.helper")
      with requests.sessions.Session() as session:
          retries = requests.adapters.Retry(
              total=Settings.max_retries_on_connection_error, backoff_factor=Settings.backoff_factor
          )
          adapter = requests.adapters.HTTPAdapter(max_retries=retries)
          session.mount("http://", adapter)
          session.mount("https://", adapter)
          try:
              r = session.get(url, allow_redirects=False, timeout=timeout)
              num_redirects = 0
              while 300 < r.status_code < 309 and num_redirects < 10:
                  num_redirects += 1
                  logger.debug("Asked to redirect({}) to: {}".format(r.status_code, r.headers["Location"]))
                  newurl = altlink(r.url, r.headers["Location"])
                  logger.info("Redirected: {}".format(urlparse(newurl).hostname))
                  r = session.get(newurl, stream=True, timeout=timeout)
          except (
              ConnectionResetError,
              requests.exceptions.ConnectionError,
              requests.exceptions.Timeout,
          ) as e:
              raise ArchiveConnectionError(f"Failure to connect to {url}: {type(e).__name__}") from e
          else:
              if r.status_code != 200:
                  msg = f"Failed to retrieve file at {url}\nServer response code: {r.status_code}, reason: {r.reason}"
                  raise ArchiveDownloadError(msg)
          result = r.text
          filename = url.split("/")[-1]
          actual_hash = hashlib.sha256(bytes(result, "utf-8")).digest()
          if expected_hash is not None and expected_hash != actual_hash:
              raise ArchiveChecksumError(
                  f"Downloaded file {filename} is corrupted! Detect checksum error.\n"
                  f"Expect {expected_hash.hex()}: {url}\n"
                  f"Actual {actual_hash.hex()}: {filename}"
              )
      return result
  
  
  def downloadBinaryFile(url: str, out: Path, hash_algo: str, exp: bytes, timeout):
      logger = getLogger("aqt.helper")
      filename = Path(url).name
      with requests.sessions.Session() as session:
          retries = requests.adapters.Retry(
              total=Settings.max_retries_on_connection_error, backoff_factor=Settings.backoff_factor
          )
          adapter = requests.adapters.HTTPAdapter(max_retries=retries)
          session.mount("http://", adapter)
          session.mount("https://", adapter)
          try:
              r = session.get(url, allow_redirects=False, stream=True, timeout=timeout)
              if 300 < r.status_code < 309:
                  logger.debug("Asked to redirect({}) to: {}".format(r.status_code, r.headers["Location"]))
                  newurl = altlink(r.url, r.headers["Location"])
                  logger.info("Redirected: {}".format(urlparse(newurl).hostname))
                  r = session.get(newurl, stream=True, timeout=timeout)
          except requests.exceptions.ConnectionError as e:
              raise ArchiveConnectionError(f"Connection error: {e.args}") from e
          except requests.exceptions.Timeout as e:
              raise ArchiveConnectionError(f"Connection timeout: {e.args}") from e
          else:
              hash = hashlib.new(hash_algo)
              try:
                  with open(out, "wb") as fd:
                      for chunk in r.iter_content(chunk_size=8196):
                          fd.write(chunk)
                          hash.update(chunk)
                      fd.flush()
              except Exception as e:
                  raise ArchiveDownloadError(f"Download of {filename} has error: {e}") from e
              if exp is not None and hash.digest() != exp:
                  raise ArchiveChecksumError(
                      f"Downloaded file {filename} is corrupted! Detect checksum error.\n"
                      f"Expect {exp.hex()}: {url}\n"
                      f"Actual {hash.digest().hex()}: {out.name}"
                  )
  
  
  def retry_on_errors(action: Callable[[], any], acceptable_errors: Tuple, num_retries: int, name: str):
      logger = getLogger("aqt.helper")
      for i in range(num_retries):
          try:
              retry_msg = f": attempt #{1 + i}" if i > 0 else ""
              logger.info(f"{name}{retry_msg}...")
              ret = action()
              if i > 0:
                  logger.info(f"Success on attempt #{1 + i}: {name}")
              return ret
          except acceptable_errors as e:
              if i < num_retries - 1:
                  continue  # just try again
              raise e from e
  
  
  def retry_on_bad_connection(function: Callable[[str], any], base_url: str):
      logger = getLogger("aqt.helper")
      fallback_url = secrets.choice(Settings.fallbacks)
      try:
          return function(base_url)
      except ArchiveConnectionError:
          logger.warning(f"Connection to '{base_url}' failed. Retrying with fallback '{fallback_url}'.")
          return function(fallback_url)
  
  
  def iter_list_reps(_list: List, num_reps: int) -> Generator:
      list_index = 0
      for i in range(num_reps):
          yield _list[list_index]
          list_index += 1
          if list_index >= len(_list):
              list_index = 0
  
  
  def get_hash(archive_path: str, algorithm: str, timeout) -> str:
      logger = getLogger("aqt.helper")
      for base_url in iter_list_reps(Settings.trusted_mirrors, Settings.max_retries_to_retrieve_hash):
          url = posixpath.join(base_url, f"{archive_path}.{algorithm}")
          logger.debug(f"Attempt to download checksum at {url}")
          try:
              r = getUrl(url, timeout)
              # sha256 & md5 files are: "some_hash archive_filename"
              return r.split(" ")[0]
          except (ArchiveConnectionError, ArchiveDownloadError):
              pass
      filename = archive_path.split("/")[-1]
      raise ChecksumDownloadFailure(
          f"Failed to download checksum for the file '{filename}' from mirrors '{Settings.trusted_mirrors}"
      )
  
  
  def altlink(url: str, alt: str):
      """
      Blacklisting redirected(alt) location based on Settings.blacklist configuration.
      When found black url, then try download a url + .meta4 that is a metalink version4
      xml file, parse it and retrieve best alternative url.
      """
      logger = getLogger("aqt.helper")
      if not any(alt.startswith(b) for b in Settings.blacklist):
          return alt
      try:
          m = _get_meta(url)
      except requests.exceptions.ConnectionError:
          logger.error("Got connection error. Fall back to recovery plan...")
          return alt
      else:
          # Expected response->'application/metalink4+xml; charset=utf-8'
          if not _check_content_type(m.headers["content-type"]):
              logger.error("Unexpected meta4 response;content-type: {}".format(m.headers["content-type"]))
              return alt
          try:
              mirror_xml = ElementTree.fromstring(m.text)
              meta_urls = {}
              for f in mirror_xml.iter("{urn:ietf:params:xml:ns:metalink}file"):
                  for u in f.iter("{urn:ietf:params:xml:ns:metalink}url"):
                      meta_urls[u.attrib["priority"]] = u.text
              mirrors = [meta_urls[i] for i in sorted(meta_urls.keys(), key=lambda x: int(x))]
          except Exception:
              exc_info = sys.exc_info()
              logger.error("Unexpected meta4 file; parse error: {}".format(exc_info[1]))
              return alt
          else:
              # Return first priority item which is not blacklist in mirrors list,
              # if not found then return alt in default
              return next(
                  filter(
                      lambda mirror: not any(mirror.startswith(b) for b in Settings.blacklist),
                      mirrors,
                  ),
                  alt,
              )
  
  
  class MyQueueListener(QueueListener):
      def __init__(self, queue):
          handlers = []
          super().__init__(queue, *handlers)
  
      def handle(self, record):
          """
          Handle a record from subprocess.
          Override logger name then handle at proper logger.
          """
          record = self.prepare(record)
          logger = getLogger("aqt.installer")
          record.name = "aqt.installer"
          logger.handle(record)
  
  
  def ssplit(data: str):
      for element in data.split(","):
          yield element.strip()
  
  
  def xml_to_modules(
      xml_text: str,
      predicate: Callable[[Element], bool],
  ) -> Dict[str, Dict[str, str]]:
      """Converts an XML document to a dict of `PackageUpdate` dicts, indexed by `Name` attribute.
      Only report elements that satisfy `predicate(element)`.
      Reports all keys available in the PackageUpdate tag as strings.
  
      :param xml_text: The entire contents of an xml file
      :param predicate: A function that decides which elements to keep or discard
      """
      try:
          parsed_xml = ElementTree.fromstring(xml_text)
      except ElementTree.ParseError as perror:
          raise ArchiveListError(f"Downloaded metadata is corrupted. {perror}") from perror
      packages = {}
      for packageupdate in parsed_xml.iter("PackageUpdate"):
          if predicate and not predicate(packageupdate):
              continue
          name = packageupdate.find("Name").text
          packages[name] = {}
          for child in packageupdate:
              packages[name][child.tag] = child.text
      return packages
  
  
  class MyConfigParser(configparser.ConfigParser):
      def getlist(self, section: str, option: str, fallback=[]) -> List[str]:
          value = self.get(section, option, fallback=None)
          if value is None:
              return fallback
          try:
              result = list(filter(None, (x.strip() for x in value.splitlines())))
          except Exception:
              result = fallback
          return result
  
      def getlistint(self, section: str, option: str, fallback=[]):
          try:
              result = [int(x) for x in self.getlist(section, option)]
          except Exception:
              result = fallback
          return result
  
  
  class SettingsClass:
      """
      Class to hold configuration and settings.
      Actual values are stored in 'settings.ini' file.
      It also holds a combinations database.
      """
  
      def __init__(self):
          self.config = MyConfigParser()
          self.configfile = os.path.join(os.path.dirname(__file__), "settings.ini")
          self.loggingconf = os.path.join(os.path.dirname(__file__), "logging.ini")
  
      def load_settings(self, file=None):
          with open(
              os.path.join(os.path.dirname(__file__), "combinations.json"),
              "r",
          ) as j:
              self._combinations = json.load(j)[0]
          if file is not None:
              if isinstance(file, str):
                  result = self.config.read(file)
                  if len(result) == 0:
                      raise IOError("Fails to load specified config file {}".format(file))
                  self.configfile = file
              else:
                  # passed through command line argparse.FileType("r")
                  self.config.read_file(file)
                  self.configfile = file
                  file.close()
          else:
              if isinstance(self.configfile, str):
                  with open(self.configfile, "r") as f:
                      self.config.read_file(f)
              else:
                  self.configfile.seek(0)
                  self.config.read_file(self.configfile)
  
      @property
      def qt_combinations(self):
          return self._combinations["qt"]
  
      @property
      def tools_combinations(self):
          return self._combinations["tools"]
  
      @property
      def available_versions(self):
          return self._combinations["versions"]
  
      @property
      def available_offline_installer_version(self):
          res = self._combinations["new_archive"]
          res.extend(self._combinations["versions"])
          return res
  
      def available_modules(self, qt_version):
          """Known module names
  
          :returns: dictionary of qt_version and module names
          :rtype: List[str]
          """
          modules = self._combinations["modules"]
          versions = qt_version.split(".")
          version = "{}.{}".format(versions[0], versions[1])
          result = None
          for record in modules:
              if record["qt_version"] == version:
                  result = record["modules"]
          return result
  
      @property
      def archive_download_location(self):
          return self.config.get("aqt", "archive_download_location", fallback=".")
  
      @property
      def always_keep_archives(self):
          return self.config.getboolean("aqt", "always_keep_archives", fallback=False)
  
      @property
      def concurrency(self):
          """concurrency configuration.
  
          :return: concurrency
          :rtype: int
          """
          return self.config.getint("aqt", "concurrency", fallback=4)
  
      @property
      def blacklist(self):
          """list of sites in a blacklist
  
          :returns: list of site URLs(scheme and host part)
          :rtype: List[str]
          """
          return self.config.getlist("mirrors", "blacklist", fallback=[])
  
      @property
      def baseurl(self):
          return self.config.get("aqt", "baseurl", fallback="https://download.qt.io")
  
      @property
      def connection_timeout(self):
          return self.config.getfloat("requests", "connection_timeout", fallback=3.5)
  
      @property
      def response_timeout(self):
          return self.config.getfloat("requests", "response_timeout", fallback=10)
  
      @property
      def max_retries(self):
          """Deprecated: please use `max_retries_on_connection_error` and `max_retries_on_checksum_error` instead!"""
          return self.config.getfloat("requests", "max_retries", fallback=5)
  
      @property
      def max_retries_on_connection_error(self):
          return self.config.getfloat("requests", "max_retries_on_connection_error", fallback=self.max_retries)
  
      @property
      def max_retries_on_checksum_error(self):
          return self.config.getint("requests", "max_retries_on_checksum_error", fallback=int(self.max_retries))
  
      @property
      def max_retries_to_retrieve_hash(self):
          return self.config.getint("requests", "max_retries_to_retrieve_hash", fallback=int(self.max_retries))
  
      @property
      def backoff_factor(self):
          return self.config.getfloat("requests", "retry_backoff", fallback=0.1)
  
      @property
      def trusted_mirrors(self):
          return self.config.getlist("mirrors", "trusted_mirrors", fallback=[self.baseurl])
  
      @property
      def fallbacks(self):
          return self.config.getlist("mirrors", "fallbacks", fallback=[])
  
      @property
      def zipcmd(self):
          return self.config.get("aqt", "7zcmd", fallback="7z")
  
      @property
      def kde_patches(self):
          return self.config.getlist("kde_patches", "patches", fallback=[])
  
      @property
      def print_stacktrace_on_error(self):
          return self.config.getboolean("aqt", "print_stacktrace_on_error", fallback=False)
  
      @property
      def min_module_size(self):
          """
          Some modules in the Qt repository contain only empty directories.
          We have found that these modules are no more than 40 bytes after decompression.
          This setting is used to filter out these empty modules in `list-*` output.
          """
          return self.config.getint("aqt", "min_module_size", fallback=41)
  
  
  Settings = SettingsClass()
  
  
  def setup_logging(env_key="LOG_CFG"):
      config = os.getenv(env_key, None)
      if config is not None and os.path.exists(config):
          logging.config.fileConfig(config)
      else:
          logging.config.fileConfig(Settings.loggingconf)
  
files.old_path: aqt/helper.py
files.new_contents: #!/usr/bin/env python3
  #
  # Copyright (C) 2018 Linus Jahn <lnj@kaidan.im>
  # Copyright (C) 2019-2021 Hiroshi Miura <miurahr@linux.com>
  # Copyright (C) 2020, Aurélien Gâteau
  #
  # Permission is hereby granted, free of charge, to any person obtaining a copy of
  # this software and associated documentation files (the "Software"), to deal in
  # the Software without restriction, including without limitation the rights to
  # use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
  # the Software, and to permit persons to whom the Software is furnished to do so,
  # subject to the following conditions:
  #
  # The above copyright notice and this permission notice shall be included in all
  # copies or substantial portions of the Software.
  #
  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
  # FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
  # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
  # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  
  import argparse
  import gc
  import multiprocessing
  import os
  import platform
  import posixpath
  import signal
  import subprocess
  import sys
  import time
  from logging import getLogger
  from logging.handlers import QueueHandler
  from pathlib import Path
  from tempfile import TemporaryDirectory
  from typing import List, Optional
  
  import aqt
  from aqt.archives import QtArchives, QtPackage, SrcDocExamplesArchives, ToolArchives
  from aqt.exceptions import (
      AqtException,
      ArchiveChecksumError,
      ArchiveDownloadError,
      ArchiveExtractionError,
      ArchiveListError,
      CliInputError,
      CliKeyboardInterrupt,
      OutOfMemory,
  )
  from aqt.helper import (
      MyQueueListener,
      Settings,
      downloadBinaryFile,
      get_hash,
      retry_on_bad_connection,
      retry_on_errors,
      setup_logging,
  )
  from aqt.metadata import ArchiveId, MetadataFactory, QtRepoProperty, SimpleSpec, Version, show_list, suggested_follow_up
  from aqt.updater import Updater
  
  try:
      import py7zr
  
      EXT7Z = False
  except ImportError:
      EXT7Z = True
  
  
  class Cli:
      """CLI main class to parse command line argument and launch proper functions."""
  
      __slot__ = ["parser", "combinations", "logger"]
  
      UNHANDLED_EXCEPTION_CODE = 254
  
      def __init__(self):
          parser = argparse.ArgumentParser(
              prog="aqt",
              description="Another unofficial Qt Installer.\naqt helps you install Qt SDK, tools, examples and others\n",
              formatter_class=argparse.RawTextHelpFormatter,
              add_help=True,
          )
          parser.add_argument(
              "-c",
              "--config",
              type=argparse.FileType("r"),
              help="Configuration ini file.",
          )
          subparsers = parser.add_subparsers(
              title="subcommands",
              description="aqt accepts several subcommands:\n"
              "install-* subcommands are commands that install components\n"
              "list-* subcommands are commands that show available components\n\n"
              "commands {install|tool|src|examples|doc} are deprecated and marked for removal\n",
              help="Please refer to each help message by using '--help' with each subcommand",
          )
          self._make_all_parsers(subparsers)
          parser.set_defaults(func=self.show_help)
          self.parser = parser
  
      def run(self, arg=None) -> int:
          args = self.parser.parse_args(arg)
          self._setup_settings(args)
          try:
              args.func(args)
              return 0
          except AqtException as e:
              self.logger.error(format(e), exc_info=Settings.print_stacktrace_on_error)
              if e.should_show_help:
                  self.show_help()
              return 1
          except Exception as e:
              # If we didn't account for it, and wrap it in an AqtException, it's a bug.
              self.logger.exception(e)  # Print stack trace
              self.logger.error(
                  f"{self._format_aqt_version()}\n"
                  f"Working dir: `{os.getcwd()}`\n"
                  f"Arguments: `{sys.argv}` Host: `{platform.uname()}`\n"
                  "===========================PLEASE FILE A BUG REPORT===========================\n"
                  "You have discovered a bug in aqt.\n"
                  "Please file a bug report at https://github.com/miurahr/aqtinstall/issues.\n"
                  "Please remember to include a copy of this program's output in your report."
              )
              return Cli.UNHANDLED_EXCEPTION_CODE
  
      def _check_tools_arg_combination(self, os_name, tool_name, arch):
          for c in Settings.tools_combinations:
              if c["os_name"] == os_name and c["tool_name"] == tool_name and c["arch"] == arch:
                  return True
          return False
  
      def _check_qt_arg_combination(self, qt_version, os_name, target, arch):
          for c in Settings.qt_combinations:
              if c["os_name"] == os_name and c["target"] == target and c["arch"] == arch:
                  return True
          return False
  
      def _check_qt_arg_versions(self, version):
          return version in Settings.available_versions
  
      def _check_qt_arg_version_offline(self, version):
          return version in Settings.available_offline_installer_version
  
      def _set_sevenzip(self, external):
          sevenzip = external
          if sevenzip is None:
              return None
  
          try:
              subprocess.run(
                  [sevenzip, "--help"],
                  stdout=subprocess.DEVNULL,
                  stderr=subprocess.DEVNULL,
              )
          except FileNotFoundError as e:
              raise CliInputError("Specified 7zip command executable does not exist: {!r}".format(sevenzip)) from e
  
          return sevenzip
  
      @staticmethod
      def _set_arch(arch: Optional[str], os_name: str, target: str, qt_version_or_spec: str) -> str:
          """Choose a default architecture, if one can be determined"""
          if arch is not None and arch != "":
              return arch
          if os_name == "linux" and target == "desktop":
              return "gcc_64"
          elif os_name == "mac" and target == "desktop":
              return "clang_64"
          elif os_name == "mac" and target == "ios":
              return "ios"
          elif target == "android":
              try:
                  if Version(qt_version_or_spec) >= Version("5.14.0"):
                      return "android"
              except ValueError:
                  pass
          raise CliInputError("Please supply a target architecture.", should_show_help=True)
  
      def _check_mirror(self, mirror):
          if mirror is None:
              pass
          elif mirror.startswith("http://") or mirror.startswith("https://") or mirror.startswith("ftp://"):
              pass
          else:
              return False
          return True
  
      def _check_modules_arg(self, qt_version, modules):
          if modules is None:
              return True
          available = Settings.available_modules(qt_version)
          if available is None:
              return False
          return all([m in available for m in modules])
  
      @staticmethod
      def _determine_qt_version(qt_version_or_spec: str, host: str, target: str, arch: str) -> Version:
          def choose_highest(x: Optional[Version], y: Optional[Version]) -> Optional[Version]:
              if x and y:
                  return max(x, y)
              return x or y
  
          def opt_version_for_spec(ext: str, _spec: SimpleSpec) -> Optional[Version]:
              try:
                  return MetadataFactory(ArchiveId("qt", host, target, ext), spec=_spec).getList().latest()
              except AqtException:
                  return None
  
          try:
              return Version(qt_version_or_spec)
          except ValueError:
              pass
          try:
              spec = SimpleSpec(qt_version_or_spec)
          except ValueError as e:
              raise CliInputError(f"Invalid version or SimpleSpec: '{qt_version_or_spec}'\n" + SimpleSpec.usage()) from e
          else:
              version: Optional[Version] = None
              for ext in QtRepoProperty.possible_extensions_for_arch(arch):
                  version = choose_highest(version, opt_version_for_spec(ext, spec))
              if not version:
                  raise CliInputError(
                      f"No versions of Qt exist for spec={spec} with host={host}, target={target}, arch={arch}"
                  )
              getLogger("aqt.installer").info(f"Resolved spec '{qt_version_or_spec}' to {version}")
              return version
  
      @staticmethod
      def choose_archive_dest(archive_dest: Optional[str], keep: bool, temp_dir: str) -> Path:
          """
          Choose archive download destination, based on context.
  
          There are three potential behaviors here:
          1. By default, return a temp directory that will be removed on program exit.
          2. If the user has asked to keep archives, but has not specified a destination,
              we return Settings.archive_download_location ("." by default).
          3. If the user has asked to keep archives and specified a destination,
              we create the destination dir if it doesn't exist, and return that directory.
          """
          if not archive_dest:
              return Path(Settings.archive_download_location if keep else temp_dir)
          dest = Path(archive_dest)
          dest.mkdir(parents=True, exist_ok=True)
          return dest
  
      def run_install_qt(self, args):
          """Run install subcommand"""
          start_time = time.perf_counter()
          self.show_aqt_version()
          if args.is_legacy:
              self._warn_on_deprecated_command("install", "install-qt")
          target: str = args.target
          os_name: str = args.host
          arch: str = self._set_arch(
              args.arch, os_name, target, getattr(args, "qt_version", getattr(args, "qt_version_spec", None))
          )
          if hasattr(args, "qt_version_spec"):
              qt_version: str = str(Cli._determine_qt_version(args.qt_version_spec, os_name, target, arch))
          else:
              qt_version: str = args.qt_version
              Cli._validate_version_str(qt_version)
          keep: bool = args.keep or Settings.always_keep_archives
          archive_dest: Optional[str] = args.archive_dest
          output_dir = args.outputdir
          if output_dir is None:
              base_dir = os.getcwd()
          else:
              base_dir = output_dir
          if args.timeout is not None:
              timeout = (args.timeout, args.timeout)
          else:
              timeout = (Settings.connection_timeout, Settings.response_timeout)
          modules = args.modules
          sevenzip = self._set_sevenzip(args.external)
          if EXT7Z and sevenzip is None:
              # override when py7zr is not exist
              sevenzip = self._set_sevenzip("7z")
          if args.base is not None:
              if not self._check_mirror(args.base):
                  raise CliInputError(
                      "The `--base` option requires a url where the path `online/qtsdkrepository` exists.",
                      should_show_help=True,
                  )
              base = args.base
          else:
              base = Settings.baseurl
          archives = args.archives
          if args.noarchives:
              if modules is None:
                  raise CliInputError("When `--noarchives` is set, the `--modules` option is mandatory.")
              if archives is not None:
                  raise CliInputError("Options `--archives` and `--noarchives` are mutually exclusive.")
          else:
              if modules is not None and archives is not None:
                  archives.append(modules)
          nopatch = args.noarchives or (archives is not None and "qtbase" not in archives)  # type: bool
          if not self._check_qt_arg_versions(qt_version):
              self.logger.warning("Specified Qt version is unknown: {}.".format(qt_version))
          if not self._check_qt_arg_combination(qt_version, os_name, target, arch):
              self.logger.warning(
                  "Specified target combination is not valid or unknown: {} {} {}".format(os_name, target, arch)
              )
          all_extra = True if modules is not None and "all" in modules else False
          if not all_extra and not self._check_modules_arg(qt_version, modules):
              self.logger.warning("Some of specified modules are unknown.")
  
          qt_archives = retry_on_bad_connection(
              lambda base_url: QtArchives(
                  os_name,
                  target,
                  qt_version,
                  arch,
                  base=base_url,
                  subarchives=archives,
                  modules=modules,
                  all_extra=all_extra,
                  is_include_base_package=not args.noarchives,
                  timeout=timeout,
              ),
              base,
          )
          target_config = qt_archives.get_target_config()
          with TemporaryDirectory() as temp_dir:
              _archive_dest = Cli.choose_archive_dest(archive_dest, keep, temp_dir)
              run_installer(qt_archives.get_packages(), base_dir, sevenzip, keep, _archive_dest)
          if not nopatch:
              Updater.update(target_config, base_dir)
          self.logger.info("Finished installation")
          self.logger.info("Time elapsed: {time:.8f} second".format(time=time.perf_counter() - start_time))
  
      def _run_src_doc_examples(self, flavor, args, cmd_name: Optional[str] = None):
          if not cmd_name:
              cmd_name = flavor
  
          self.show_aqt_version()
          if args.is_legacy:
              self._warn_on_deprecated_command(old_name=cmd_name, new_name=f"install-{cmd_name}")
          elif getattr(args, "target", None) is not None:
              self._warn_on_deprecated_parameter("target", args.target)
          target = "desktop"  # The only valid target for src/doc/examples is "desktop"
          os_name = args.host
          if hasattr(args, "qt_version_spec"):
              qt_version = str(Cli._determine_qt_version(args.qt_version_spec, os_name, target, arch=""))
          else:
              qt_version = args.qt_version
              Cli._validate_version_str(qt_version)
          output_dir = args.outputdir
          if output_dir is None:
              base_dir = os.getcwd()
          else:
              base_dir = output_dir
          keep: bool = args.keep or Settings.always_keep_archives
          archive_dest: Optional[str] = args.archive_dest
          if args.base is not None:
              base = args.base
          else:
              base = Settings.baseurl
          if args.timeout is not None:
              timeout = (args.timeout, args.timeout)
          else:
              timeout = (Settings.connection_timeout, Settings.response_timeout)
          sevenzip = self._set_sevenzip(args.external)
          if EXT7Z and sevenzip is None:
              # override when py7zr is not exist
              sevenzip = self._set_sevenzip(Settings.zipcmd)
          modules = getattr(args, "modules", None)  # `--modules` is invalid for `install-src`
          archives = args.archives
          all_extra = True if modules is not None and "all" in modules else False
          if not self._check_qt_arg_versions(qt_version):
              self.logger.warning("Specified Qt version is unknown: {}.".format(qt_version))
  
          srcdocexamples_archives: SrcDocExamplesArchives = retry_on_bad_connection(
              lambda base_url: SrcDocExamplesArchives(
                  flavor,
                  os_name,
                  target,
                  qt_version,
                  base=base_url,
                  subarchives=archives,
                  modules=modules,
                  all_extra=all_extra,
                  timeout=timeout,
              ),
              base,
          )
          with TemporaryDirectory() as temp_dir:
              _archive_dest = Cli.choose_archive_dest(archive_dest, keep, temp_dir)
              run_installer(srcdocexamples_archives.get_packages(), base_dir, sevenzip, keep, _archive_dest)
          self.logger.info("Finished installation")
  
      def run_install_src(self, args):
          """Run src subcommand"""
          if not hasattr(args, "qt_version"):
              args.qt_version = str(Cli._determine_qt_version(args.qt_version_spec, args.host, args.target, arch=""))
          if args.kde and args.qt_version != "5.15.2":
              raise CliInputError("KDE patch: unsupported version!!")
          start_time = time.perf_counter()
          self._run_src_doc_examples("src", args)
          if args.kde:
              if args.outputdir is None:
                  target_dir = os.path.join(os.getcwd(), args.qt_version, "Src")
              else:
                  target_dir = os.path.join(args.outputdir, args.qt_version, "Src")
              Updater.patch_kde(target_dir)
          self.logger.info("Time elapsed: {time:.8f} second".format(time=time.perf_counter() - start_time))
  
      def run_install_example(self, args):
          """Run example subcommand"""
          start_time = time.perf_counter()
          self._run_src_doc_examples("examples", args, cmd_name="example")
          self.logger.info("Time elapsed: {time:.8f} second".format(time=time.perf_counter() - start_time))
  
      def run_install_doc(self, args):
          """Run doc subcommand"""
          start_time = time.perf_counter()
          self._run_src_doc_examples("doc", args)
          self.logger.info("Time elapsed: {time:.8f} second".format(time=time.perf_counter() - start_time))
  
      def run_install_tool(self, args):
          """Run tool subcommand"""
          start_time = time.perf_counter()
          self.show_aqt_version()
          if args.is_legacy:
              self._warn_on_deprecated_command("tool", "install-tool")
          tool_name = args.tool_name  # such as tools_openssl_x64
          os_name = args.host  # windows, linux and mac
          target = "desktop" if args.is_legacy else args.target  # desktop, android and ios
          output_dir = args.outputdir
          if output_dir is None:
              base_dir = os.getcwd()
          else:
              base_dir = output_dir
          sevenzip = self._set_sevenzip(args.external)
          if EXT7Z and sevenzip is None:
              # override when py7zr is not exist
              sevenzip = self._set_sevenzip(Settings.zipcmd)
          version = getattr(args, "version", None)
          if version is not None:
              Cli._validate_version_str(version, allow_minus=True)
          keep: bool = args.keep or Settings.always_keep_archives
          archive_dest: Optional[str] = args.archive_dest
          if args.base is not None:
              base = args.base
          else:
              base = Settings.baseurl
          if args.timeout is not None:
              timeout = (args.timeout, args.timeout)
          else:
              timeout = (Settings.connection_timeout, Settings.response_timeout)
          if args.tool_variant is None:
              archive_id = ArchiveId("tools", os_name, target, "")
              meta = MetadataFactory(archive_id, is_latest_version=True, tool_name=tool_name)
              try:
                  archs = meta.getList()
              except ArchiveDownloadError as e:
                  msg = f"Failed to locate XML data for the tool '{tool_name}'."
                  raise ArchiveListError(msg, suggested_action=suggested_follow_up(meta)) from e
  
          else:
              archs = [args.tool_variant]
  
          for arch in archs:
              if not self._check_tools_arg_combination(os_name, tool_name, arch):
                  self.logger.warning("Specified target combination is not valid: {} {} {}".format(os_name, tool_name, arch))
  
              tool_archives: ToolArchives = retry_on_bad_connection(
                  lambda base_url: ToolArchives(
                      os_name=os_name,
                      tool_name=tool_name,
                      target=target,
                      base=base_url,
                      version_str=version,
                      arch=arch,
                      timeout=timeout,
                  ),
                  base,
              )
              with TemporaryDirectory() as temp_dir:
                  _archive_dest = Cli.choose_archive_dest(archive_dest, keep, temp_dir)
                  run_installer(tool_archives.get_packages(), base_dir, sevenzip, keep, _archive_dest)
          self.logger.info("Finished installation")
          self.logger.info("Time elapsed: {time:.8f} second".format(time=time.perf_counter() - start_time))
  
      def run_list_qt(self, args: argparse.ArgumentParser):
          """Print versions of Qt, extensions, modules, architectures"""
  
          if not args.target:
              print(" ".join(ArchiveId.TARGETS_FOR_HOST[args.host]))
              return
          if args.target not in ArchiveId.TARGETS_FOR_HOST[args.host]:
              raise CliInputError("'{0.target}' is not a valid target for host '{0.host}'".format(args))
          if args.modules:
              modules_ver, modules_query = args.modules[0], tuple(args.modules)
          else:
              modules_ver, modules_query = None, None
  
          for version_str in (modules_ver, args.extensions, args.arch, args.archives[0] if args.archives else None):
              Cli._validate_version_str(version_str, allow_latest=True, allow_empty=True)
  
          spec = None
          try:
              if args.spec is not None:
                  spec = SimpleSpec(args.spec)
          except ValueError as e:
              raise CliInputError(f"Invalid version specification: '{args.spec}'.\n" + SimpleSpec.usage()) from e
  
          meta = MetadataFactory(
              archive_id=ArchiveId(
                  "qt",
                  args.host,
                  args.target,
                  args.extension if args.extension else "",
              ),
              spec=spec,
              is_latest_version=args.latest_version,
              modules_query=modules_query,
              extensions_ver=args.extensions,
              architectures_ver=args.arch,
              archives_query=args.archives,
          )
          show_list(meta)
  
      def run_list_tool(self, args: argparse.ArgumentParser):
          """Print tools"""
  
          if not args.target:
              print(" ".join(ArchiveId.TARGETS_FOR_HOST[args.host]))
              return
          if args.target not in ArchiveId.TARGETS_FOR_HOST[args.host]:
              raise CliInputError("'{0.target}' is not a valid target for host '{0.host}'".format(args))
  
          meta = MetadataFactory(
              archive_id=ArchiveId("tools", args.host, args.target),
              tool_name=args.tool_name,
              is_long_listing=args.long,
          )
          show_list(meta)
  
      def run_list_src_doc_examples(self, args: argparse.ArgumentParser, cmd_type: str):
          target = "desktop"  # The only valid target for src/doc/examples is "desktop"
          version = Cli._determine_qt_version(args.qt_version_spec, args.host, target, arch="")
          is_fetch_modules: bool = getattr(args, "modules", False)
          meta = MetadataFactory(
              archive_id=ArchiveId("qt", args.host, target, "src_doc_examples"),
              src_doc_examples_query=(cmd_type, version, is_fetch_modules),
          )
          show_list(meta)
  
      def show_help(self, args=None):
          """Display help message"""
          self.parser.print_help()
  
      def _format_aqt_version(self) -> str:
          py_version = platform.python_version()
          py_impl = platform.python_implementation()
          py_build = platform.python_compiler()
          return f"aqtinstall(aqt) v{aqt.__version__} on Python {py_version} [{py_impl} {py_build}]"
  
      def show_aqt_version(self, args=None):
          """Display version information"""
          self.logger.info(self._format_aqt_version())
  
      def _set_install_qt_parser(self, install_qt_parser, *, is_legacy: bool):
          install_qt_parser.set_defaults(func=self.run_install_qt, is_legacy=is_legacy)
          self._set_common_arguments(install_qt_parser, is_legacy=is_legacy)
          self._set_common_options(install_qt_parser)
          install_qt_parser.add_argument(
              "arch",
              nargs="?",
              help="\ntarget linux/desktop: gcc_64, wasm_32"
              "\ntarget mac/desktop:   clang_64, wasm_32"
              "\ntarget mac/ios:       ios"
              "\nwindows/desktop:      win64_msvc2019_64, win32_msvc2019"
              "\n                      win64_msvc2017_64, win32_msvc2017"
              "\n                      win64_msvc2015_64, win32_msvc2015"
              "\n                      win64_mingw81, win32_mingw81"
              "\n                      win64_mingw73, win32_mingw73"
              "\n                      win32_mingw53"
              "\n                      wasm_32"
              "\nwindows/winrt:        win64_msvc2019_winrt_x64, win64_msvc2019_winrt_x86"
              "\n                      win64_msvc2017_winrt_x64, win64_msvc2017_winrt_x86"
              "\n                      win64_msvc2019_winrt_armv7"
              "\n                      win64_msvc2017_winrt_armv7"
              "\nandroid:              Qt 5.14:          android (optional)"
              "\n                      Qt 5.13 or below: android_x86_64, android_arm64_v8a"
              "\n                                        android_x86, android_armv7",
          )
          self._set_module_options(install_qt_parser)
          self._set_archive_options(install_qt_parser)
          install_qt_parser.add_argument(
              "--noarchives",
              action="store_true",
              help="No base packages; allow mod amendment with --modules option.",
          )
  
      def _set_install_tool_parser(self, install_tool_parser, *, is_legacy: bool):
          install_tool_parser.set_defaults(func=self.run_install_tool, is_legacy=is_legacy)
          install_tool_parser.add_argument("host", choices=["linux", "mac", "windows"], help="host os name")
          if not is_legacy:
              install_tool_parser.add_argument(
                  "target",
                  default=None,
                  choices=["desktop", "winrt", "android", "ios"],
                  help="Target SDK.",
              )
          install_tool_parser.add_argument("tool_name", help="Name of tool such as tools_ifw, tools_mingw")
          if is_legacy:
              install_tool_parser.add_argument("version", help="Version of tool variant")
  
          tool_variant_opts = {} if is_legacy else {"nargs": "?", "default": None}
          install_tool_parser.add_argument(
              "tool_variant",
              **tool_variant_opts,
              help="Name of tool variant, such as qt.tools.ifw.41. "
              "Please use 'aqt list-tool' to list acceptable values for this parameter.",
          )
          self._set_common_options(install_tool_parser)
  
      def _warn_on_deprecated_command(self, old_name: str, new_name: str):
          self.logger.warning(
              f"Warning: The command '{old_name}' is deprecated and marked for removal in a future version of aqt.\n"
              f"In the future, please use the command '{new_name}' instead."
          )
  
      def _warn_on_deprecated_parameter(self, parameter_name: str, value: str):
          self.logger.warning(
              f"Warning: The parameter '{parameter_name}' with value '{value}' is deprecated and marked for "
              f"removal in a future version of aqt.\n"
              f"In the future, please omit this parameter."
          )
  
      def _make_all_parsers(self, subparsers: argparse._SubParsersAction):
          deprecated_msg = "This command is deprecated and marked for removal in a future version of aqt."
  
          def make_parser_it(cmd: str, desc: str, is_legacy: bool, set_parser_cmd, formatter_class):
              description = f"{desc} {deprecated_msg}" if is_legacy else desc
              kwargs = {"formatter_class": formatter_class} if formatter_class else {}
              p = subparsers.add_parser(cmd, description=description, **kwargs)
              set_parser_cmd(p, is_legacy=is_legacy)
  
          def make_parser_sde(cmd: str, desc: str, is_legacy: bool, action, is_add_kde: bool, is_add_modules: bool = True):
              description = f"{desc} {deprecated_msg}" if is_legacy else desc
              parser = subparsers.add_parser(cmd, description=description)
              parser.set_defaults(func=action, is_legacy=is_legacy)
              self._set_common_arguments(parser, is_legacy=is_legacy, is_target_deprecated=True)
              self._set_common_options(parser)
              if is_add_modules:
                  self._set_module_options(parser)
              self._set_archive_options(parser)
              if is_add_kde:
                  parser.add_argument("--kde", action="store_true", help="patching with KDE patch kit.")
  
          def make_parser_list_sde(cmd: str, desc: str, cmd_type: str):
              parser = subparsers.add_parser(cmd, description=desc)
              parser.add_argument("host", choices=["linux", "mac", "windows"], help="host os name")
              parser.add_argument(
                  "qt_version_spec",
                  metavar="(VERSION | SPECIFICATION)",
                  help='Qt version in the format of "5.X.Y" or SimpleSpec like "5.X" or "<6.X"',
              )
              parser.set_defaults(func=lambda args: self.run_list_src_doc_examples(args, cmd_type))
  
              if cmd_type != "src":
                  parser.add_argument("-m", "--modules", action="store_true", help="Print list of available modules")
  
          make_parser_it("install-qt", "Install Qt.", False, self._set_install_qt_parser, argparse.RawTextHelpFormatter)
          make_parser_it("install-tool", "Install tools.", False, self._set_install_tool_parser, None)
          make_parser_sde("install-doc", "Install documentation.", False, self.run_install_doc, False)
          make_parser_sde("install-example", "Install examples.", False, self.run_install_example, False)
          make_parser_sde("install-src", "Install source.", False, self.run_install_src, True, is_add_modules=False)
  
          self._make_list_qt_parser(subparsers)
          self._make_list_tool_parser(subparsers)
          make_parser_list_sde("list-doc", "List documentation archives available (use with install-doc)", "doc")
          make_parser_list_sde("list-example", "List example archives available (use with install-example)", "examples")
          make_parser_list_sde("list-src", "List source archives available (use with install-src)", "src")
  
          make_parser_it("install", "Install Qt.", True, self._set_install_qt_parser, argparse.RawTextHelpFormatter)
          make_parser_it("tool", "Install tools.", True, self._set_install_tool_parser, None)
          make_parser_sde("doc", "Install documentation.", True, self.run_install_doc, False)
          make_parser_sde("examples", "Install examples.", True, self.run_install_example, False)
          make_parser_sde("src", "Install source.", True, self.run_install_src, True)
  
          self._make_common_parsers(subparsers)
  
      def _make_list_qt_parser(self, subparsers: argparse._SubParsersAction):
          """Creates a subparser that works with the MetadataFactory, and adds it to the `subparsers` parameter"""
          list_parser: argparse.ArgumentParser = subparsers.add_parser(
              "list-qt",
              formatter_class=argparse.RawDescriptionHelpFormatter,
              epilog="Examples:\n"
              "$ aqt list-qt mac                                                # print all targets for Mac OS\n"
              "$ aqt list-qt mac desktop                                        # print all versions of Qt 5\n"
              "$ aqt list-qt mac desktop --extension wasm                       # print all wasm versions of Qt 5\n"
              '$ aqt list-qt mac desktop --spec "5.9"                           # print all versions of Qt 5.9\n'
              '$ aqt list-qt mac desktop --spec "5.9" --latest-version          # print latest Qt 5.9\n'
              "$ aqt list-qt mac desktop --modules 5.12.0 clang_64              # print modules for 5.12.0\n"
              "$ aqt list-qt mac desktop --spec 5.9 --modules latest clang_64   # print modules for latest 5.9\n"
              "$ aqt list-qt mac desktop --extensions 5.9.0                     # print choices for --extension flag\n"
              "$ aqt list-qt mac desktop --arch 5.9.9                           # print architectures for 5.9.9/mac/desktop\n"
              "$ aqt list-qt mac desktop --arch latest                          # print architectures for the latest Qt 5\n"
              "$ aqt list-qt mac desktop --archives 5.9.0 clang_64              # list archives in base Qt installation\n"
              "$ aqt list-qt mac desktop --archives 5.14.0 clang_64 debug_info  # list archives in debug_info module\n",
          )
          list_parser.add_argument("host", choices=["linux", "mac", "windows"], help="host os name")
          list_parser.add_argument(
              "target",
              nargs="?",
              default=None,
              choices=["desktop", "winrt", "android", "ios"],
              help="Target SDK. When omitted, this prints all the targets available for a host OS.",
          )
          list_parser.add_argument(
              "--extension",
              choices=ArchiveId.ALL_EXTENSIONS,
              help="Extension of packages to list. "
              "Use the `--extensions` flag to list all relevant options for a host/target.",
          )
          list_parser.add_argument(
              "--spec",
              type=str,
              metavar="SPECIFICATION",
              help="Filter output so that only versions that match the specification are printed. "
              'IE: `aqt list-qt windows desktop --spec "5.12"` prints all versions beginning with 5.12',
          )
          output_modifier_exclusive_group = list_parser.add_mutually_exclusive_group()
          output_modifier_exclusive_group.add_argument(
              "--modules",
              type=str,
              nargs=2,
              metavar=("(VERSION | latest)", "ARCHITECTURE"),
              help='First arg: Qt version in the format of "5.X.Y", or the keyword "latest". '
              'Second arg: an architecture, which may be printed with the "--arch" flag. '
              "When set, this prints all the modules available for either Qt 5.X.Y or the latest version of Qt.",
          )
          output_modifier_exclusive_group.add_argument(
              "--extensions",
              type=str,
              metavar="(VERSION | latest)",
              help='Qt version in the format of "5.X.Y", or the keyword "latest". '
              "When set, this prints all valid arguments for the `--extension` flag "
              "for either Qt 5.X.Y or the latest version of Qt.",
          )
          output_modifier_exclusive_group.add_argument(
              "--arch",
              type=str,
              metavar="(VERSION | latest)",
              help='Qt version in the format of "5.X.Y", or the keyword "latest". '
              "When set, this prints all architectures available for either Qt 5.X.Y or the latest version of Qt.",
          )
          output_modifier_exclusive_group.add_argument(
              "--latest-version",
              action="store_true",
              help="print only the newest version available",
          )
          output_modifier_exclusive_group.add_argument(
              "--archives",
              type=str,
              nargs="+",
              help="print the archives available for Qt base or modules. "
              "If two arguments are provided, the first two arguments must be 'VERSION | latest' and "
              "'ARCHITECTURE', and this command will print all archives associated with the base Qt package. "
              "If more than two arguments are provided, the remaining arguments will be interpreted as modules, "
              "and this command will print all archives associated with those modules. "
              "At least two arguments are required.",
          )
          list_parser.set_defaults(func=self.run_list_qt)
  
      def _make_list_tool_parser(self, subparsers: argparse._SubParsersAction):
          """Creates a subparser that works with the MetadataFactory, and adds it to the `subparsers` parameter"""
          list_parser: argparse.ArgumentParser = subparsers.add_parser(
              "list-tool",
              formatter_class=argparse.RawDescriptionHelpFormatter,
              epilog="Examples:\n"
              "$ aqt list-tool mac desktop                 # print all tools for mac desktop\n"
              "$ aqt list-tool mac desktop tools_ifw       # print all tool variant names for QtIFW\n"
              "$ aqt list-tool mac desktop ifw             # print all tool variant names for QtIFW\n"
              "$ aqt list-tool mac desktop -l tools_ifw    # print tool variant names with metadata for QtIFW\n"
              "$ aqt list-tool mac desktop -l ifw          # print tool variant names with metadata for QtIFW\n",
          )
          list_parser.add_argument("host", choices=["linux", "mac", "windows"], help="host os name")
          list_parser.add_argument(
              "target",
              nargs="?",
              default=None,
              choices=["desktop", "winrt", "android", "ios"],
              help="Target SDK. When omitted, this prints all the targets available for a host OS.",
          )
          list_parser.add_argument(
              "tool_name",
              nargs="?",
              default=None,
              help='Name of a tool, ie "tools_mingw" or "tools_ifw". '
              "When omitted, this prints all the tool names available for a host OS/target SDK combination. "
              "When present, this prints all the tool variant names available for this tool. ",
          )
          list_parser.add_argument(
              "-l",
              "--long",
              action="store_true",
              help="Long display: shows a table of metadata associated with each tool variant. "
              "On narrow terminals, it displays tool variant names, versions, and release dates. "
              "On terminals wider than 95 characters, it also displays descriptions of each tool.",
          )
          list_parser.set_defaults(func=self.run_list_tool)
  
      def _make_common_parsers(self, subparsers: argparse._SubParsersAction):
          help_parser = subparsers.add_parser("help")
          help_parser.set_defaults(func=self.show_help)
          #
          version_parser = subparsers.add_parser("version")
          version_parser.set_defaults(func=self.show_aqt_version)
  
      def _set_common_options(self, subparser):
          subparser.add_argument(
              "-O",
              "--outputdir",
              nargs="?",
              help="Target output directory(default current directory)",
          )
          subparser.add_argument(
              "-b",
              "--base",
              nargs="?",
              help="Specify mirror base url such as http://mirrors.ocf.berkeley.edu/qt/, " "where 'online' folder exist.",
          )
          subparser.add_argument(
              "--timeout",
              nargs="?",
              type=float,
              help="Specify connection timeout for download site.(default: 5 sec)",
          )
          subparser.add_argument("-E", "--external", nargs="?", help="Specify external 7zip command path.")
          subparser.add_argument("--internal", action="store_true", help="Use internal extractor.")
          subparser.add_argument(
              "-k",
              "--keep",
              action="store_true",
              help="Keep downloaded archive when specified, otherwise remove after install",
          )
          subparser.add_argument(
              "-d",
              "--archive-dest",
              type=str,
              default=None,
              help="Set the destination path for downloaded archives (temp directory by default).",
          )
  
      def _set_module_options(self, subparser):
          subparser.add_argument("-m", "--modules", nargs="*", help="Specify extra modules to install")
  
      def _set_archive_options(self, subparser):
          subparser.add_argument(
              "--archives",
              nargs="*",
              help="Specify subset of archives to install. Affects the base module and the debug_info module. "
              "(Default: all archives).",
          )
  
      def _set_common_arguments(self, subparser, *, is_legacy: bool, is_target_deprecated: bool = False):
          """
          Legacy commands require that the version comes before host and target.
          Non-legacy commands require that the host and target are before the version.
          install-src/doc/example commands do not require a "target" argument anymore, as of 11/22/2021
          """
          if is_legacy:
              subparser.add_argument("qt_version", help='Qt version in the format of "5.X.Y"')
          subparser.add_argument("host", choices=["linux", "mac", "windows"], help="host os name")
          if is_target_deprecated:
              subparser.add_argument(
                  "target",
                  choices=["desktop", "winrt", "android", "ios"],
                  nargs="?",
                  help="Ignored. This parameter is deprecated and marked for removal in a future release. "
                  "It is present here for backwards compatibility.",
              )
          else:
              subparser.add_argument("target", choices=["desktop", "winrt", "android", "ios"], help="target sdk")
          if not is_legacy:
              subparser.add_argument(
                  "qt_version_spec",
                  metavar="(VERSION | SPECIFICATION)",
                  help='Qt version in the format of "5.X.Y" or SimpleSpec like "5.X" or "<6.X"',
              )
  
      def _setup_settings(self, args=None):
          # setup logging
          setup_logging()
          self.logger = getLogger("aqt.main")
          # setup settings
          if args is not None and args.config is not None:
              Settings.load_settings(args.config)
          else:
              config = os.getenv("AQT_CONFIG", None)
              if config is not None and os.path.exists(config):
                  Settings.load_settings(config)
                  self.logger.debug("Load configuration from {}".format(config))
              else:
                  Settings.load_settings()
  
      @staticmethod
      def _validate_version_str(
          version_str: str, *, allow_latest: bool = False, allow_empty: bool = False, allow_minus: bool = False
      ) -> None:
          """
          Raise CliInputError if the version is not an acceptable Version.
  
          :param version_str: The version string to check.
          :param allow_latest: If true, the string "latest" is acceptable.
          :param allow_empty: If true, the empty string is acceptable.
          :param allow_minus: If true, everything after the first '-' in the version will be ignored.
                              This allows acceptance of versions like "1.2.3-0-202101020304"
          """
          if (allow_latest and version_str == "latest") or (allow_empty and not version_str):
              return
          try:
              if "-" in version_str and allow_minus:
                  version_str = version_str[: version_str.find("-")]
              Version(version_str)
          except ValueError as e:
              raise CliInputError(f"Invalid version: '{version_str}'! Please use the form '5.X.Y'.") from e
  
  
  def is_64bit() -> bool:
      """check if running platform is 64bit python."""
      return sys.maxsize > 1 << 32
  
  
  def run_installer(archives: List[QtPackage], base_dir: str, sevenzip: Optional[str], keep: bool, archive_dest: Path):
      queue = multiprocessing.Manager().Queue(-1)
      listener = MyQueueListener(queue)
      listener.start()
      #
      tasks = []
      for arc in archives:
          tasks.append((arc, base_dir, sevenzip, queue, archive_dest, keep))
      ctx = multiprocessing.get_context("spawn")
      if is_64bit():
          pool = ctx.Pool(Settings.concurrency, init_worker_sh, (), 4)
      else:
          pool = ctx.Pool(Settings.concurrency, init_worker_sh, (), 1)
  
      def close_worker_pool_on_exception(exception: BaseException):
          logger = getLogger("aqt.installer")
          logger.warning(f"Caught {exception.__class__.__name__}, terminating installer workers")
          pool.terminate()
          pool.join()
  
      try:
          pool.starmap(installer, tasks)
          pool.close()
          pool.join()
      except KeyboardInterrupt as e:
          close_worker_pool_on_exception(e)
          raise CliKeyboardInterrupt("Installer halted by keyboard interrupt.") from e
      except MemoryError as e:
          close_worker_pool_on_exception(e)
          alt_extractor_msg = (
              "Please try using the '--external' flag to specify an alternate 7z extraction tool "
              "(see https://aqtinstall.readthedocs.io/en/latest/cli.html#cmdoption-list-tool-external)"
          )
          if Settings.concurrency > 1:
              docs_url = "https://aqtinstall.readthedocs.io/en/stable/configuration.html#configuration"
              raise OutOfMemory(
                  "Out of memory when downloading and extracting archives in parallel.",
                  suggested_action=[f"Please reduce your 'concurrency' setting (see {docs_url})", alt_extractor_msg],
              ) from e
          raise OutOfMemory(
              "Out of memory when downloading and extracting archives.",
              suggested_action=["Please free up more memory.", alt_extractor_msg],
          )
      except Exception as e:
          close_worker_pool_on_exception(e)
          raise e from e
      finally:
          # all done, close logging service for sub-processes
          listener.enqueue_sentinel()
          listener.stop()
  
  
  def init_worker_sh():
      signal.signal(signal.SIGINT, signal.SIG_IGN)
  
  
  def installer(
      qt_package: QtPackage,
      base_dir: str,
      command: Optional[str],
      queue: multiprocessing.Queue,
      archive_dest: Path,
      keep: bool = False,
      response_timeout: Optional[int] = None,
  ):
      """
      Installer function to download archive files and extract it.
      It is called through multiprocessing.Pool()
      """
      name = qt_package.name
      base_url = qt_package.base_url
      archive: Path = archive_dest / qt_package.archive
      start_time = time.perf_counter()
      # set defaults
      Settings.load_settings()
      # set logging
      setup_logging()  # XXX: why need to load again?
      qh = QueueHandler(queue)
      logger = getLogger()
      for handler in logger.handlers:
          handler.close()
          logger.removeHandler(handler)
      logger.addHandler(qh)
      #
      if response_timeout is None:
          timeout = (Settings.connection_timeout, Settings.response_timeout)
      else:
          timeout = (Settings.connection_timeout, response_timeout)
      hash = get_hash(qt_package.archive_path, algorithm="sha256", timeout=timeout)
  
      def download_bin(_base_url):
          url = posixpath.join(_base_url, qt_package.archive_path)
          logger.debug("Download URL: {}".format(url))
          return downloadBinaryFile(url, archive, "sha256", hash, timeout)
  
      retry_on_errors(
          action=lambda: retry_on_bad_connection(download_bin, base_url),
          acceptable_errors=(ArchiveChecksumError,),
          num_retries=Settings.max_retries_on_checksum_error,
          name=f"Downloading {name}",
      )
      gc.collect()
      if command is None:
          with py7zr.SevenZipFile(archive, "r") as szf:
              szf.extractall(path=base_dir)
      else:
          if base_dir is not None:
              command_args = [
                  command,
                  "x",
                  "-aoa",
                  "-bd",
                  "-y",
                  "-o{}".format(base_dir),
                  str(archive),
              ]
          else:
              command_args = [command, "x", "-aoa", "-bd", "-y", str(archive)]
          try:
              proc = subprocess.run(command_args, stdout=subprocess.PIPE, check=True)
              logger.debug(proc.stdout)
          except subprocess.CalledProcessError as cpe:
              msg = "\n".join(filter(None, [f"Extraction error: {cpe.returncode}", cpe.stdout, cpe.stderr]))
              raise ArchiveExtractionError(msg) from cpe
      if not keep:
          os.unlink(archive)
      logger.info("Finished installation of {} in {:.8f}".format(archive.name, time.perf_counter() - start_time))
      gc.collect()
      qh.flush()
      qh.close()
      logger.removeHandler(qh)
  
files.new_path: aqt/installer.py
files.old_contents: #!/usr/bin/env python3
  #
  # Copyright (C) 2018 Linus Jahn <lnj@kaidan.im>
  # Copyright (C) 2019-2021 Hiroshi Miura <miurahr@linux.com>
  # Copyright (C) 2020, Aurélien Gâteau
  #
  # Permission is hereby granted, free of charge, to any person obtaining a copy of
  # this software and associated documentation files (the "Software"), to deal in
  # the Software without restriction, including without limitation the rights to
  # use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
  # the Software, and to permit persons to whom the Software is furnished to do so,
  # subject to the following conditions:
  #
  # The above copyright notice and this permission notice shall be included in all
  # copies or substantial portions of the Software.
  #
  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
  # FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
  # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
  # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  
  import argparse
  import binascii
  import gc
  import multiprocessing
  import os
  import platform
  import posixpath
  import signal
  import subprocess
  import sys
  import time
  from logging import getLogger
  from logging.handlers import QueueHandler
  from pathlib import Path
  from tempfile import TemporaryDirectory
  from typing import List, Optional
  
  import aqt
  from aqt.archives import QtArchives, QtPackage, SrcDocExamplesArchives, ToolArchives
  from aqt.exceptions import (
      AqtException,
      ArchiveChecksumError,
      ArchiveDownloadError,
      ArchiveExtractionError,
      ArchiveListError,
      CliInputError,
      CliKeyboardInterrupt,
      OutOfMemory,
  )
  from aqt.helper import (
      MyQueueListener,
      Settings,
      downloadBinaryFile,
      get_hash,
      retry_on_bad_connection,
      retry_on_errors,
      setup_logging,
  )
  from aqt.metadata import ArchiveId, MetadataFactory, QtRepoProperty, SimpleSpec, Version, show_list, suggested_follow_up
  from aqt.updater import Updater
  
  try:
      import py7zr
  
      EXT7Z = False
  except ImportError:
      EXT7Z = True
  
  
  class Cli:
      """CLI main class to parse command line argument and launch proper functions."""
  
      __slot__ = ["parser", "combinations", "logger"]
  
      UNHANDLED_EXCEPTION_CODE = 254
  
      def __init__(self):
          parser = argparse.ArgumentParser(
              prog="aqt",
              description="Another unofficial Qt Installer.\naqt helps you install Qt SDK, tools, examples and others\n",
              formatter_class=argparse.RawTextHelpFormatter,
              add_help=True,
          )
          parser.add_argument(
              "-c",
              "--config",
              type=argparse.FileType("r"),
              help="Configuration ini file.",
          )
          subparsers = parser.add_subparsers(
              title="subcommands",
              description="aqt accepts several subcommands:\n"
              "install-* subcommands are commands that install components\n"
              "list-* subcommands are commands that show available components\n\n"
              "commands {install|tool|src|examples|doc} are deprecated and marked for removal\n",
              help="Please refer to each help message by using '--help' with each subcommand",
          )
          self._make_all_parsers(subparsers)
          parser.set_defaults(func=self.show_help)
          self.parser = parser
  
      def run(self, arg=None) -> int:
          args = self.parser.parse_args(arg)
          self._setup_settings(args)
          try:
              args.func(args)
              return 0
          except AqtException as e:
              self.logger.error(format(e), exc_info=Settings.print_stacktrace_on_error)
              if e.should_show_help:
                  self.show_help()
              return 1
          except Exception as e:
              # If we didn't account for it, and wrap it in an AqtException, it's a bug.
              self.logger.exception(e)  # Print stack trace
              self.logger.error(
                  f"{self._format_aqt_version()}\n"
                  f"Working dir: `{os.getcwd()}`\n"
                  f"Arguments: `{sys.argv}` Host: `{platform.uname()}`\n"
                  "===========================PLEASE FILE A BUG REPORT===========================\n"
                  "You have discovered a bug in aqt.\n"
                  "Please file a bug report at https://github.com/miurahr/aqtinstall/issues.\n"
                  "Please remember to include a copy of this program's output in your report."
              )
              return Cli.UNHANDLED_EXCEPTION_CODE
  
      def _check_tools_arg_combination(self, os_name, tool_name, arch):
          for c in Settings.tools_combinations:
              if c["os_name"] == os_name and c["tool_name"] == tool_name and c["arch"] == arch:
                  return True
          return False
  
      def _check_qt_arg_combination(self, qt_version, os_name, target, arch):
          for c in Settings.qt_combinations:
              if c["os_name"] == os_name and c["target"] == target and c["arch"] == arch:
                  return True
          return False
  
      def _check_qt_arg_versions(self, version):
          return version in Settings.available_versions
  
      def _check_qt_arg_version_offline(self, version):
          return version in Settings.available_offline_installer_version
  
      def _set_sevenzip(self, external):
          sevenzip = external
          if sevenzip is None:
              return None
  
          try:
              subprocess.run(
                  [sevenzip, "--help"],
                  stdout=subprocess.DEVNULL,
                  stderr=subprocess.DEVNULL,
              )
          except FileNotFoundError as e:
              raise CliInputError("Specified 7zip command executable does not exist: {!r}".format(sevenzip)) from e
  
          return sevenzip
  
      @staticmethod
      def _set_arch(arch: Optional[str], os_name: str, target: str, qt_version_or_spec: str) -> str:
          """Choose a default architecture, if one can be determined"""
          if arch is not None and arch != "":
              return arch
          if os_name == "linux" and target == "desktop":
              return "gcc_64"
          elif os_name == "mac" and target == "desktop":
              return "clang_64"
          elif os_name == "mac" and target == "ios":
              return "ios"
          elif target == "android":
              try:
                  if Version(qt_version_or_spec) >= Version("5.14.0"):
                      return "android"
              except ValueError:
                  pass
          raise CliInputError("Please supply a target architecture.", should_show_help=True)
  
      def _check_mirror(self, mirror):
          if mirror is None:
              pass
          elif mirror.startswith("http://") or mirror.startswith("https://") or mirror.startswith("ftp://"):
              pass
          else:
              return False
          return True
  
      def _check_modules_arg(self, qt_version, modules):
          if modules is None:
              return True
          available = Settings.available_modules(qt_version)
          if available is None:
              return False
          return all([m in available for m in modules])
  
      @staticmethod
      def _determine_qt_version(qt_version_or_spec: str, host: str, target: str, arch: str) -> Version:
          def choose_highest(x: Optional[Version], y: Optional[Version]) -> Optional[Version]:
              if x and y:
                  return max(x, y)
              return x or y
  
          def opt_version_for_spec(ext: str, _spec: SimpleSpec) -> Optional[Version]:
              try:
                  return MetadataFactory(ArchiveId("qt", host, target, ext), spec=_spec).getList().latest()
              except AqtException:
                  return None
  
          try:
              return Version(qt_version_or_spec)
          except ValueError:
              pass
          try:
              spec = SimpleSpec(qt_version_or_spec)
          except ValueError as e:
              raise CliInputError(f"Invalid version or SimpleSpec: '{qt_version_or_spec}'\n" + SimpleSpec.usage()) from e
          else:
              version: Optional[Version] = None
              for ext in QtRepoProperty.possible_extensions_for_arch(arch):
                  version = choose_highest(version, opt_version_for_spec(ext, spec))
              if not version:
                  raise CliInputError(
                      f"No versions of Qt exist for spec={spec} with host={host}, target={target}, arch={arch}"
                  )
              getLogger("aqt.installer").info(f"Resolved spec '{qt_version_or_spec}' to {version}")
              return version
  
      @staticmethod
      def choose_archive_dest(archive_dest: Optional[str], keep: bool, temp_dir: str) -> Path:
          """
          Choose archive download destination, based on context.
  
          There are three potential behaviors here:
          1. By default, return a temp directory that will be removed on program exit.
          2. If the user has asked to keep archives, but has not specified a destination,
              we return Settings.archive_download_location ("." by default).
          3. If the user has asked to keep archives and specified a destination,
              we create the destination dir if it doesn't exist, and return that directory.
          """
          if not archive_dest:
              return Path(Settings.archive_download_location if keep else temp_dir)
          dest = Path(archive_dest)
          dest.mkdir(parents=True, exist_ok=True)
          return dest
  
      def run_install_qt(self, args):
          """Run install subcommand"""
          start_time = time.perf_counter()
          self.show_aqt_version()
          if args.is_legacy:
              self._warn_on_deprecated_command("install", "install-qt")
          target: str = args.target
          os_name: str = args.host
          arch: str = self._set_arch(
              args.arch, os_name, target, getattr(args, "qt_version", getattr(args, "qt_version_spec", None))
          )
          if hasattr(args, "qt_version_spec"):
              qt_version: str = str(Cli._determine_qt_version(args.qt_version_spec, os_name, target, arch))
          else:
              qt_version: str = args.qt_version
              Cli._validate_version_str(qt_version)
          keep: bool = args.keep or Settings.always_keep_archives
          archive_dest: Optional[str] = args.archive_dest
          output_dir = args.outputdir
          if output_dir is None:
              base_dir = os.getcwd()
          else:
              base_dir = output_dir
          if args.timeout is not None:
              timeout = (args.timeout, args.timeout)
          else:
              timeout = (Settings.connection_timeout, Settings.response_timeout)
          modules = args.modules
          sevenzip = self._set_sevenzip(args.external)
          if EXT7Z and sevenzip is None:
              # override when py7zr is not exist
              sevenzip = self._set_sevenzip("7z")
          if args.base is not None:
              if not self._check_mirror(args.base):
                  raise CliInputError(
                      "The `--base` option requires a url where the path `online/qtsdkrepository` exists.",
                      should_show_help=True,
                  )
              base = args.base
          else:
              base = Settings.baseurl
          archives = args.archives
          if args.noarchives:
              if modules is None:
                  raise CliInputError("When `--noarchives` is set, the `--modules` option is mandatory.")
              if archives is not None:
                  raise CliInputError("Options `--archives` and `--noarchives` are mutually exclusive.")
          else:
              if modules is not None and archives is not None:
                  archives.append(modules)
          nopatch = args.noarchives or (archives is not None and "qtbase" not in archives)  # type: bool
          if not self._check_qt_arg_versions(qt_version):
              self.logger.warning("Specified Qt version is unknown: {}.".format(qt_version))
          if not self._check_qt_arg_combination(qt_version, os_name, target, arch):
              self.logger.warning(
                  "Specified target combination is not valid or unknown: {} {} {}".format(os_name, target, arch)
              )
          all_extra = True if modules is not None and "all" in modules else False
          if not all_extra and not self._check_modules_arg(qt_version, modules):
              self.logger.warning("Some of specified modules are unknown.")
  
          qt_archives = retry_on_bad_connection(
              lambda base_url: QtArchives(
                  os_name,
                  target,
                  qt_version,
                  arch,
                  base=base_url,
                  subarchives=archives,
                  modules=modules,
                  all_extra=all_extra,
                  is_include_base_package=not args.noarchives,
                  timeout=timeout,
              ),
              base,
          )
          target_config = qt_archives.get_target_config()
          with TemporaryDirectory() as temp_dir:
              _archive_dest = Cli.choose_archive_dest(archive_dest, keep, temp_dir)
              run_installer(qt_archives.get_packages(), base_dir, sevenzip, keep, _archive_dest)
          if not nopatch:
              Updater.update(target_config, base_dir)
          self.logger.info("Finished installation")
          self.logger.info("Time elapsed: {time:.8f} second".format(time=time.perf_counter() - start_time))
  
      def _run_src_doc_examples(self, flavor, args, cmd_name: Optional[str] = None):
          if not cmd_name:
              cmd_name = flavor
  
          self.show_aqt_version()
          if args.is_legacy:
              self._warn_on_deprecated_command(old_name=cmd_name, new_name=f"install-{cmd_name}")
          elif getattr(args, "target", None) is not None:
              self._warn_on_deprecated_parameter("target", args.target)
          target = "desktop"  # The only valid target for src/doc/examples is "desktop"
          os_name = args.host
          if hasattr(args, "qt_version_spec"):
              qt_version = str(Cli._determine_qt_version(args.qt_version_spec, os_name, target, arch=""))
          else:
              qt_version = args.qt_version
              Cli._validate_version_str(qt_version)
          output_dir = args.outputdir
          if output_dir is None:
              base_dir = os.getcwd()
          else:
              base_dir = output_dir
          keep: bool = args.keep or Settings.always_keep_archives
          archive_dest: Optional[str] = args.archive_dest
          if args.base is not None:
              base = args.base
          else:
              base = Settings.baseurl
          if args.timeout is not None:
              timeout = (args.timeout, args.timeout)
          else:
              timeout = (Settings.connection_timeout, Settings.response_timeout)
          sevenzip = self._set_sevenzip(args.external)
          if EXT7Z and sevenzip is None:
              # override when py7zr is not exist
              sevenzip = self._set_sevenzip(Settings.zipcmd)
          modules = getattr(args, "modules", None)  # `--modules` is invalid for `install-src`
          archives = args.archives
          all_extra = True if modules is not None and "all" in modules else False
          if not self._check_qt_arg_versions(qt_version):
              self.logger.warning("Specified Qt version is unknown: {}.".format(qt_version))
  
          srcdocexamples_archives: SrcDocExamplesArchives = retry_on_bad_connection(
              lambda base_url: SrcDocExamplesArchives(
                  flavor,
                  os_name,
                  target,
                  qt_version,
                  base=base_url,
                  subarchives=archives,
                  modules=modules,
                  all_extra=all_extra,
                  timeout=timeout,
              ),
              base,
          )
          with TemporaryDirectory() as temp_dir:
              _archive_dest = Cli.choose_archive_dest(archive_dest, keep, temp_dir)
              run_installer(srcdocexamples_archives.get_packages(), base_dir, sevenzip, keep, _archive_dest)
          self.logger.info("Finished installation")
  
      def run_install_src(self, args):
          """Run src subcommand"""
          if not hasattr(args, "qt_version"):
              args.qt_version = str(Cli._determine_qt_version(args.qt_version_spec, args.host, args.target, arch=""))
          if args.kde and args.qt_version != "5.15.2":
              raise CliInputError("KDE patch: unsupported version!!")
          start_time = time.perf_counter()
          self._run_src_doc_examples("src", args)
          if args.kde:
              if args.outputdir is None:
                  target_dir = os.path.join(os.getcwd(), args.qt_version, "Src")
              else:
                  target_dir = os.path.join(args.outputdir, args.qt_version, "Src")
              Updater.patch_kde(target_dir)
          self.logger.info("Time elapsed: {time:.8f} second".format(time=time.perf_counter() - start_time))
  
      def run_install_example(self, args):
          """Run example subcommand"""
          start_time = time.perf_counter()
          self._run_src_doc_examples("examples", args, cmd_name="example")
          self.logger.info("Time elapsed: {time:.8f} second".format(time=time.perf_counter() - start_time))
  
      def run_install_doc(self, args):
          """Run doc subcommand"""
          start_time = time.perf_counter()
          self._run_src_doc_examples("doc", args)
          self.logger.info("Time elapsed: {time:.8f} second".format(time=time.perf_counter() - start_time))
  
      def run_install_tool(self, args):
          """Run tool subcommand"""
          start_time = time.perf_counter()
          self.show_aqt_version()
          if args.is_legacy:
              self._warn_on_deprecated_command("tool", "install-tool")
          tool_name = args.tool_name  # such as tools_openssl_x64
          os_name = args.host  # windows, linux and mac
          target = "desktop" if args.is_legacy else args.target  # desktop, android and ios
          output_dir = args.outputdir
          if output_dir is None:
              base_dir = os.getcwd()
          else:
              base_dir = output_dir
          sevenzip = self._set_sevenzip(args.external)
          if EXT7Z and sevenzip is None:
              # override when py7zr is not exist
              sevenzip = self._set_sevenzip(Settings.zipcmd)
          version = getattr(args, "version", None)
          if version is not None:
              Cli._validate_version_str(version, allow_minus=True)
          keep: bool = args.keep or Settings.always_keep_archives
          archive_dest: Optional[str] = args.archive_dest
          if args.base is not None:
              base = args.base
          else:
              base = Settings.baseurl
          if args.timeout is not None:
              timeout = (args.timeout, args.timeout)
          else:
              timeout = (Settings.connection_timeout, Settings.response_timeout)
          if args.tool_variant is None:
              archive_id = ArchiveId("tools", os_name, target, "")
              meta = MetadataFactory(archive_id, is_latest_version=True, tool_name=tool_name)
              try:
                  archs = meta.getList()
              except ArchiveDownloadError as e:
                  msg = f"Failed to locate XML data for the tool '{tool_name}'."
                  raise ArchiveListError(msg, suggested_action=suggested_follow_up(meta)) from e
  
          else:
              archs = [args.tool_variant]
  
          for arch in archs:
              if not self._check_tools_arg_combination(os_name, tool_name, arch):
                  self.logger.warning("Specified target combination is not valid: {} {} {}".format(os_name, tool_name, arch))
  
              tool_archives: ToolArchives = retry_on_bad_connection(
                  lambda base_url: ToolArchives(
                      os_name=os_name,
                      tool_name=tool_name,
                      target=target,
                      base=base_url,
                      version_str=version,
                      arch=arch,
                      timeout=timeout,
                  ),
                  base,
              )
              with TemporaryDirectory() as temp_dir:
                  _archive_dest = Cli.choose_archive_dest(archive_dest, keep, temp_dir)
                  run_installer(tool_archives.get_packages(), base_dir, sevenzip, keep, _archive_dest)
          self.logger.info("Finished installation")
          self.logger.info("Time elapsed: {time:.8f} second".format(time=time.perf_counter() - start_time))
  
      def run_list_qt(self, args: argparse.ArgumentParser):
          """Print versions of Qt, extensions, modules, architectures"""
  
          if not args.target:
              print(" ".join(ArchiveId.TARGETS_FOR_HOST[args.host]))
              return
          if args.target not in ArchiveId.TARGETS_FOR_HOST[args.host]:
              raise CliInputError("'{0.target}' is not a valid target for host '{0.host}'".format(args))
          if args.modules:
              modules_ver, modules_query = args.modules[0], tuple(args.modules)
          else:
              modules_ver, modules_query = None, None
  
          for version_str in (modules_ver, args.extensions, args.arch, args.archives[0] if args.archives else None):
              Cli._validate_version_str(version_str, allow_latest=True, allow_empty=True)
  
          spec = None
          try:
              if args.spec is not None:
                  spec = SimpleSpec(args.spec)
          except ValueError as e:
              raise CliInputError(f"Invalid version specification: '{args.spec}'.\n" + SimpleSpec.usage()) from e
  
          meta = MetadataFactory(
              archive_id=ArchiveId(
                  "qt",
                  args.host,
                  args.target,
                  args.extension if args.extension else "",
              ),
              spec=spec,
              is_latest_version=args.latest_version,
              modules_query=modules_query,
              extensions_ver=args.extensions,
              architectures_ver=args.arch,
              archives_query=args.archives,
          )
          show_list(meta)
  
      def run_list_tool(self, args: argparse.ArgumentParser):
          """Print tools"""
  
          if not args.target:
              print(" ".join(ArchiveId.TARGETS_FOR_HOST[args.host]))
              return
          if args.target not in ArchiveId.TARGETS_FOR_HOST[args.host]:
              raise CliInputError("'{0.target}' is not a valid target for host '{0.host}'".format(args))
  
          meta = MetadataFactory(
              archive_id=ArchiveId("tools", args.host, args.target),
              tool_name=args.tool_name,
              is_long_listing=args.long,
          )
          show_list(meta)
  
      def run_list_src_doc_examples(self, args: argparse.ArgumentParser, cmd_type: str):
          target = "desktop"  # The only valid target for src/doc/examples is "desktop"
          version = Cli._determine_qt_version(args.qt_version_spec, args.host, target, arch="")
          is_fetch_modules: bool = getattr(args, "modules", False)
          meta = MetadataFactory(
              archive_id=ArchiveId("qt", args.host, target, "src_doc_examples"),
              src_doc_examples_query=(cmd_type, version, is_fetch_modules),
          )
          show_list(meta)
  
      def show_help(self, args=None):
          """Display help message"""
          self.parser.print_help()
  
      def _format_aqt_version(self) -> str:
          py_version = platform.python_version()
          py_impl = platform.python_implementation()
          py_build = platform.python_compiler()
          return f"aqtinstall(aqt) v{aqt.__version__} on Python {py_version} [{py_impl} {py_build}]"
  
      def show_aqt_version(self, args=None):
          """Display version information"""
          self.logger.info(self._format_aqt_version())
  
      def _set_install_qt_parser(self, install_qt_parser, *, is_legacy: bool):
          install_qt_parser.set_defaults(func=self.run_install_qt, is_legacy=is_legacy)
          self._set_common_arguments(install_qt_parser, is_legacy=is_legacy)
          self._set_common_options(install_qt_parser)
          install_qt_parser.add_argument(
              "arch",
              nargs="?",
              help="\ntarget linux/desktop: gcc_64, wasm_32"
              "\ntarget mac/desktop:   clang_64, wasm_32"
              "\ntarget mac/ios:       ios"
              "\nwindows/desktop:      win64_msvc2019_64, win32_msvc2019"
              "\n                      win64_msvc2017_64, win32_msvc2017"
              "\n                      win64_msvc2015_64, win32_msvc2015"
              "\n                      win64_mingw81, win32_mingw81"
              "\n                      win64_mingw73, win32_mingw73"
              "\n                      win32_mingw53"
              "\n                      wasm_32"
              "\nwindows/winrt:        win64_msvc2019_winrt_x64, win64_msvc2019_winrt_x86"
              "\n                      win64_msvc2017_winrt_x64, win64_msvc2017_winrt_x86"
              "\n                      win64_msvc2019_winrt_armv7"
              "\n                      win64_msvc2017_winrt_armv7"
              "\nandroid:              Qt 5.14:          android (optional)"
              "\n                      Qt 5.13 or below: android_x86_64, android_arm64_v8a"
              "\n                                        android_x86, android_armv7",
          )
          self._set_module_options(install_qt_parser)
          self._set_archive_options(install_qt_parser)
          install_qt_parser.add_argument(
              "--noarchives",
              action="store_true",
              help="No base packages; allow mod amendment with --modules option.",
          )
  
      def _set_install_tool_parser(self, install_tool_parser, *, is_legacy: bool):
          install_tool_parser.set_defaults(func=self.run_install_tool, is_legacy=is_legacy)
          install_tool_parser.add_argument("host", choices=["linux", "mac", "windows"], help="host os name")
          if not is_legacy:
              install_tool_parser.add_argument(
                  "target",
                  default=None,
                  choices=["desktop", "winrt", "android", "ios"],
                  help="Target SDK.",
              )
          install_tool_parser.add_argument("tool_name", help="Name of tool such as tools_ifw, tools_mingw")
          if is_legacy:
              install_tool_parser.add_argument("version", help="Version of tool variant")
  
          tool_variant_opts = {} if is_legacy else {"nargs": "?", "default": None}
          install_tool_parser.add_argument(
              "tool_variant",
              **tool_variant_opts,
              help="Name of tool variant, such as qt.tools.ifw.41. "
              "Please use 'aqt list-tool' to list acceptable values for this parameter.",
          )
          self._set_common_options(install_tool_parser)
  
      def _warn_on_deprecated_command(self, old_name: str, new_name: str):
          self.logger.warning(
              f"Warning: The command '{old_name}' is deprecated and marked for removal in a future version of aqt.\n"
              f"In the future, please use the command '{new_name}' instead."
          )
  
      def _warn_on_deprecated_parameter(self, parameter_name: str, value: str):
          self.logger.warning(
              f"Warning: The parameter '{parameter_name}' with value '{value}' is deprecated and marked for "
              f"removal in a future version of aqt.\n"
              f"In the future, please omit this parameter."
          )
  
      def _make_all_parsers(self, subparsers: argparse._SubParsersAction):
          deprecated_msg = "This command is deprecated and marked for removal in a future version of aqt."
  
          def make_parser_it(cmd: str, desc: str, is_legacy: bool, set_parser_cmd, formatter_class):
              description = f"{desc} {deprecated_msg}" if is_legacy else desc
              kwargs = {"formatter_class": formatter_class} if formatter_class else {}
              p = subparsers.add_parser(cmd, description=description, **kwargs)
              set_parser_cmd(p, is_legacy=is_legacy)
  
          def make_parser_sde(cmd: str, desc: str, is_legacy: bool, action, is_add_kde: bool, is_add_modules: bool = True):
              description = f"{desc} {deprecated_msg}" if is_legacy else desc
              parser = subparsers.add_parser(cmd, description=description)
              parser.set_defaults(func=action, is_legacy=is_legacy)
              self._set_common_arguments(parser, is_legacy=is_legacy, is_target_deprecated=True)
              self._set_common_options(parser)
              if is_add_modules:
                  self._set_module_options(parser)
              self._set_archive_options(parser)
              if is_add_kde:
                  parser.add_argument("--kde", action="store_true", help="patching with KDE patch kit.")
  
          def make_parser_list_sde(cmd: str, desc: str, cmd_type: str):
              parser = subparsers.add_parser(cmd, description=desc)
              parser.add_argument("host", choices=["linux", "mac", "windows"], help="host os name")
              parser.add_argument(
                  "qt_version_spec",
                  metavar="(VERSION | SPECIFICATION)",
                  help='Qt version in the format of "5.X.Y" or SimpleSpec like "5.X" or "<6.X"',
              )
              parser.set_defaults(func=lambda args: self.run_list_src_doc_examples(args, cmd_type))
  
              if cmd_type != "src":
                  parser.add_argument("-m", "--modules", action="store_true", help="Print list of available modules")
  
          make_parser_it("install-qt", "Install Qt.", False, self._set_install_qt_parser, argparse.RawTextHelpFormatter)
          make_parser_it("install-tool", "Install tools.", False, self._set_install_tool_parser, None)
          make_parser_sde("install-doc", "Install documentation.", False, self.run_install_doc, False)
          make_parser_sde("install-example", "Install examples.", False, self.run_install_example, False)
          make_parser_sde("install-src", "Install source.", False, self.run_install_src, True, is_add_modules=False)
  
          self._make_list_qt_parser(subparsers)
          self._make_list_tool_parser(subparsers)
          make_parser_list_sde("list-doc", "List documentation archives available (use with install-doc)", "doc")
          make_parser_list_sde("list-example", "List example archives available (use with install-example)", "examples")
          make_parser_list_sde("list-src", "List source archives available (use with install-src)", "src")
  
          make_parser_it("install", "Install Qt.", True, self._set_install_qt_parser, argparse.RawTextHelpFormatter)
          make_parser_it("tool", "Install tools.", True, self._set_install_tool_parser, None)
          make_parser_sde("doc", "Install documentation.", True, self.run_install_doc, False)
          make_parser_sde("examples", "Install examples.", True, self.run_install_example, False)
          make_parser_sde("src", "Install source.", True, self.run_install_src, True)
  
          self._make_common_parsers(subparsers)
  
      def _make_list_qt_parser(self, subparsers: argparse._SubParsersAction):
          """Creates a subparser that works with the MetadataFactory, and adds it to the `subparsers` parameter"""
          list_parser: argparse.ArgumentParser = subparsers.add_parser(
              "list-qt",
              formatter_class=argparse.RawDescriptionHelpFormatter,
              epilog="Examples:\n"
              "$ aqt list-qt mac                                                # print all targets for Mac OS\n"
              "$ aqt list-qt mac desktop                                        # print all versions of Qt 5\n"
              "$ aqt list-qt mac desktop --extension wasm                       # print all wasm versions of Qt 5\n"
              '$ aqt list-qt mac desktop --spec "5.9"                           # print all versions of Qt 5.9\n'
              '$ aqt list-qt mac desktop --spec "5.9" --latest-version          # print latest Qt 5.9\n'
              "$ aqt list-qt mac desktop --modules 5.12.0 clang_64              # print modules for 5.12.0\n"
              "$ aqt list-qt mac desktop --spec 5.9 --modules latest clang_64   # print modules for latest 5.9\n"
              "$ aqt list-qt mac desktop --extensions 5.9.0                     # print choices for --extension flag\n"
              "$ aqt list-qt mac desktop --arch 5.9.9                           # print architectures for 5.9.9/mac/desktop\n"
              "$ aqt list-qt mac desktop --arch latest                          # print architectures for the latest Qt 5\n"
              "$ aqt list-qt mac desktop --archives 5.9.0 clang_64              # list archives in base Qt installation\n"
              "$ aqt list-qt mac desktop --archives 5.14.0 clang_64 debug_info  # list archives in debug_info module\n",
          )
          list_parser.add_argument("host", choices=["linux", "mac", "windows"], help="host os name")
          list_parser.add_argument(
              "target",
              nargs="?",
              default=None,
              choices=["desktop", "winrt", "android", "ios"],
              help="Target SDK. When omitted, this prints all the targets available for a host OS.",
          )
          list_parser.add_argument(
              "--extension",
              choices=ArchiveId.ALL_EXTENSIONS,
              help="Extension of packages to list. "
              "Use the `--extensions` flag to list all relevant options for a host/target.",
          )
          list_parser.add_argument(
              "--spec",
              type=str,
              metavar="SPECIFICATION",
              help="Filter output so that only versions that match the specification are printed. "
              'IE: `aqt list-qt windows desktop --spec "5.12"` prints all versions beginning with 5.12',
          )
          output_modifier_exclusive_group = list_parser.add_mutually_exclusive_group()
          output_modifier_exclusive_group.add_argument(
              "--modules",
              type=str,
              nargs=2,
              metavar=("(VERSION | latest)", "ARCHITECTURE"),
              help='First arg: Qt version in the format of "5.X.Y", or the keyword "latest". '
              'Second arg: an architecture, which may be printed with the "--arch" flag. '
              "When set, this prints all the modules available for either Qt 5.X.Y or the latest version of Qt.",
          )
          output_modifier_exclusive_group.add_argument(
              "--extensions",
              type=str,
              metavar="(VERSION | latest)",
              help='Qt version in the format of "5.X.Y", or the keyword "latest". '
              "When set, this prints all valid arguments for the `--extension` flag "
              "for either Qt 5.X.Y or the latest version of Qt.",
          )
          output_modifier_exclusive_group.add_argument(
              "--arch",
              type=str,
              metavar="(VERSION | latest)",
              help='Qt version in the format of "5.X.Y", or the keyword "latest". '
              "When set, this prints all architectures available for either Qt 5.X.Y or the latest version of Qt.",
          )
          output_modifier_exclusive_group.add_argument(
              "--latest-version",
              action="store_true",
              help="print only the newest version available",
          )
          output_modifier_exclusive_group.add_argument(
              "--archives",
              type=str,
              nargs="+",
              help="print the archives available for Qt base or modules. "
              "If two arguments are provided, the first two arguments must be 'VERSION | latest' and "
              "'ARCHITECTURE', and this command will print all archives associated with the base Qt package. "
              "If more than two arguments are provided, the remaining arguments will be interpreted as modules, "
              "and this command will print all archives associated with those modules. "
              "At least two arguments are required.",
          )
          list_parser.set_defaults(func=self.run_list_qt)
  
      def _make_list_tool_parser(self, subparsers: argparse._SubParsersAction):
          """Creates a subparser that works with the MetadataFactory, and adds it to the `subparsers` parameter"""
          list_parser: argparse.ArgumentParser = subparsers.add_parser(
              "list-tool",
              formatter_class=argparse.RawDescriptionHelpFormatter,
              epilog="Examples:\n"
              "$ aqt list-tool mac desktop                 # print all tools for mac desktop\n"
              "$ aqt list-tool mac desktop tools_ifw       # print all tool variant names for QtIFW\n"
              "$ aqt list-tool mac desktop ifw             # print all tool variant names for QtIFW\n"
              "$ aqt list-tool mac desktop -l tools_ifw    # print tool variant names with metadata for QtIFW\n"
              "$ aqt list-tool mac desktop -l ifw          # print tool variant names with metadata for QtIFW\n",
          )
          list_parser.add_argument("host", choices=["linux", "mac", "windows"], help="host os name")
          list_parser.add_argument(
              "target",
              nargs="?",
              default=None,
              choices=["desktop", "winrt", "android", "ios"],
              help="Target SDK. When omitted, this prints all the targets available for a host OS.",
          )
          list_parser.add_argument(
              "tool_name",
              nargs="?",
              default=None,
              help='Name of a tool, ie "tools_mingw" or "tools_ifw". '
              "When omitted, this prints all the tool names available for a host OS/target SDK combination. "
              "When present, this prints all the tool variant names available for this tool. ",
          )
          list_parser.add_argument(
              "-l",
              "--long",
              action="store_true",
              help="Long display: shows a table of metadata associated with each tool variant. "
              "On narrow terminals, it displays tool variant names, versions, and release dates. "
              "On terminals wider than 95 characters, it also displays descriptions of each tool.",
          )
          list_parser.set_defaults(func=self.run_list_tool)
  
      def _make_common_parsers(self, subparsers: argparse._SubParsersAction):
          help_parser = subparsers.add_parser("help")
          help_parser.set_defaults(func=self.show_help)
          #
          version_parser = subparsers.add_parser("version")
          version_parser.set_defaults(func=self.show_aqt_version)
  
      def _set_common_options(self, subparser):
          subparser.add_argument(
              "-O",
              "--outputdir",
              nargs="?",
              help="Target output directory(default current directory)",
          )
          subparser.add_argument(
              "-b",
              "--base",
              nargs="?",
              help="Specify mirror base url such as http://mirrors.ocf.berkeley.edu/qt/, " "where 'online' folder exist.",
          )
          subparser.add_argument(
              "--timeout",
              nargs="?",
              type=float,
              help="Specify connection timeout for download site.(default: 5 sec)",
          )
          subparser.add_argument("-E", "--external", nargs="?", help="Specify external 7zip command path.")
          subparser.add_argument("--internal", action="store_true", help="Use internal extractor.")
          subparser.add_argument(
              "-k",
              "--keep",
              action="store_true",
              help="Keep downloaded archive when specified, otherwise remove after install",
          )
          subparser.add_argument(
              "-d",
              "--archive-dest",
              type=str,
              default=None,
              help="Set the destination path for downloaded archives (temp directory by default).",
          )
  
      def _set_module_options(self, subparser):
          subparser.add_argument("-m", "--modules", nargs="*", help="Specify extra modules to install")
  
      def _set_archive_options(self, subparser):
          subparser.add_argument(
              "--archives",
              nargs="*",
              help="Specify subset of archives to install. Affects the base module and the debug_info module. "
              "(Default: all archives).",
          )
  
      def _set_common_arguments(self, subparser, *, is_legacy: bool, is_target_deprecated: bool = False):
          """
          Legacy commands require that the version comes before host and target.
          Non-legacy commands require that the host and target are before the version.
          install-src/doc/example commands do not require a "target" argument anymore, as of 11/22/2021
          """
          if is_legacy:
              subparser.add_argument("qt_version", help='Qt version in the format of "5.X.Y"')
          subparser.add_argument("host", choices=["linux", "mac", "windows"], help="host os name")
          if is_target_deprecated:
              subparser.add_argument(
                  "target",
                  choices=["desktop", "winrt", "android", "ios"],
                  nargs="?",
                  help="Ignored. This parameter is deprecated and marked for removal in a future release. "
                  "It is present here for backwards compatibility.",
              )
          else:
              subparser.add_argument("target", choices=["desktop", "winrt", "android", "ios"], help="target sdk")
          if not is_legacy:
              subparser.add_argument(
                  "qt_version_spec",
                  metavar="(VERSION | SPECIFICATION)",
                  help='Qt version in the format of "5.X.Y" or SimpleSpec like "5.X" or "<6.X"',
              )
  
      def _setup_settings(self, args=None):
          # setup logging
          setup_logging()
          self.logger = getLogger("aqt.main")
          # setup settings
          if args is not None and args.config is not None:
              Settings.load_settings(args.config)
          else:
              config = os.getenv("AQT_CONFIG", None)
              if config is not None and os.path.exists(config):
                  Settings.load_settings(config)
                  self.logger.debug("Load configuration from {}".format(config))
              else:
                  Settings.load_settings()
  
      @staticmethod
      def _validate_version_str(
          version_str: str, *, allow_latest: bool = False, allow_empty: bool = False, allow_minus: bool = False
      ) -> None:
          """
          Raise CliInputError if the version is not an acceptable Version.
  
          :param version_str: The version string to check.
          :param allow_latest: If true, the string "latest" is acceptable.
          :param allow_empty: If true, the empty string is acceptable.
          :param allow_minus: If true, everything after the first '-' in the version will be ignored.
                              This allows acceptance of versions like "1.2.3-0-202101020304"
          """
          if (allow_latest and version_str == "latest") or (allow_empty and not version_str):
              return
          try:
              if "-" in version_str and allow_minus:
                  version_str = version_str[: version_str.find("-")]
              Version(version_str)
          except ValueError as e:
              raise CliInputError(f"Invalid version: '{version_str}'! Please use the form '5.X.Y'.") from e
  
  
  def is_64bit() -> bool:
      """check if running platform is 64bit python."""
      return sys.maxsize > 1 << 32
  
  
  def run_installer(archives: List[QtPackage], base_dir: str, sevenzip: Optional[str], keep: bool, archive_dest: Path):
      queue = multiprocessing.Manager().Queue(-1)
      listener = MyQueueListener(queue)
      listener.start()
      #
      tasks = []
      for arc in archives:
          tasks.append((arc, base_dir, sevenzip, queue, archive_dest, keep))
      ctx = multiprocessing.get_context("spawn")
      if is_64bit():
          pool = ctx.Pool(Settings.concurrency, init_worker_sh, (), 4)
      else:
          pool = ctx.Pool(Settings.concurrency, init_worker_sh, (), 1)
  
      def close_worker_pool_on_exception(exception: BaseException):
          logger = getLogger("aqt.installer")
          logger.warning(f"Caught {exception.__class__.__name__}, terminating installer workers")
          pool.terminate()
          pool.join()
  
      try:
          pool.starmap(installer, tasks)
          pool.close()
          pool.join()
      except KeyboardInterrupt as e:
          close_worker_pool_on_exception(e)
          raise CliKeyboardInterrupt("Installer halted by keyboard interrupt.") from e
      except MemoryError as e:
          close_worker_pool_on_exception(e)
          alt_extractor_msg = (
              "Please try using the '--external' flag to specify an alternate 7z extraction tool "
              "(see https://aqtinstall.readthedocs.io/en/latest/cli.html#cmdoption-list-tool-external)"
          )
          if Settings.concurrency > 1:
              docs_url = "https://aqtinstall.readthedocs.io/en/stable/configuration.html#configuration"
              raise OutOfMemory(
                  "Out of memory when downloading and extracting archives in parallel.",
                  suggested_action=[f"Please reduce your 'concurrency' setting (see {docs_url})", alt_extractor_msg],
              ) from e
          raise OutOfMemory(
              "Out of memory when downloading and extracting archives.",
              suggested_action=["Please free up more memory.", alt_extractor_msg],
          )
      except Exception as e:
          close_worker_pool_on_exception(e)
          raise e from e
      finally:
          # all done, close logging service for sub-processes
          listener.enqueue_sentinel()
          listener.stop()
  
  
  def init_worker_sh():
      signal.signal(signal.SIGINT, signal.SIG_IGN)
  
  
  def installer(
      qt_package: QtPackage,
      base_dir: str,
      command: Optional[str],
      queue: multiprocessing.Queue,
      archive_dest: Path,
      keep: bool = False,
      response_timeout: Optional[int] = None,
  ):
      """
      Installer function to download archive files and extract it.
      It is called through multiprocessing.Pool()
      """
      name = qt_package.name
      base_url = qt_package.base_url
      archive: Path = archive_dest / qt_package.archive
      start_time = time.perf_counter()
      # set defaults
      Settings.load_settings()
      # set logging
      setup_logging()  # XXX: why need to load again?
      qh = QueueHandler(queue)
      logger = getLogger()
      for handler in logger.handlers:
          handler.close()
          logger.removeHandler(handler)
      logger.addHandler(qh)
      #
      if response_timeout is None:
          timeout = (Settings.connection_timeout, Settings.response_timeout)
      else:
          timeout = (Settings.connection_timeout, response_timeout)
      hash = binascii.unhexlify(get_hash(qt_package.archive_path, algorithm="sha256", timeout=timeout))
  
      def download_bin(_base_url):
          url = posixpath.join(_base_url, qt_package.archive_path)
          logger.debug("Download URL: {}".format(url))
          return downloadBinaryFile(url, archive, "sha256", hash, timeout)
  
      retry_on_errors(
          action=lambda: retry_on_bad_connection(download_bin, base_url),
          acceptable_errors=(ArchiveChecksumError,),
          num_retries=Settings.max_retries_on_checksum_error,
          name=f"Downloading {name}",
      )
      gc.collect()
      if command is None:
          with py7zr.SevenZipFile(archive, "r") as szf:
              szf.extractall(path=base_dir)
      else:
          if base_dir is not None:
              command_args = [
                  command,
                  "x",
                  "-aoa",
                  "-bd",
                  "-y",
                  "-o{}".format(base_dir),
                  str(archive),
              ]
          else:
              command_args = [command, "x", "-aoa", "-bd", "-y", str(archive)]
          try:
              proc = subprocess.run(command_args, stdout=subprocess.PIPE, check=True)
              logger.debug(proc.stdout)
          except subprocess.CalledProcessError as cpe:
              msg = "\n".join(filter(None, [f"Extraction error: {cpe.returncode}", cpe.stdout, cpe.stderr]))
              raise ArchiveExtractionError(msg) from cpe
      if not keep:
          os.unlink(archive)
      logger.info("Finished installation of {} in {:.8f}".format(archive.name, time.perf_counter() - start_time))
      gc.collect()
      qh.flush()
      qh.close()
      logger.removeHandler(qh)
  
files.old_path: aqt/installer.py
files.new_contents: #!/usr/bin/env python
  #
  # Copyright (C) 2021 David Dalcino
  #
  # Permission is hereby granted, free of charge, to any person obtaining a copy of
  # this software and associated documentation files (the "Software"), to deal in
  # the Software without restriction, including without limitation the rights to
  # use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
  # the Software, and to permit persons to whom the Software is furnished to do so,
  # subject to the following conditions:
  #
  # The above copyright notice and this permission notice shall be included in all
  # copies or substantial portions of the Software.
  #
  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
  # FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
  # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
  # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  import itertools
  import operator
  import posixpath
  import re
  import secrets as random
  import shutil
  from logging import getLogger
  from typing import Callable, Dict, Generator, Iterable, Iterator, List, Optional, Tuple, Union
  from xml.etree.ElementTree import Element
  
  import bs4
  from semantic_version import SimpleSpec as SemanticSimpleSpec
  from semantic_version import Version as SemanticVersion
  from texttable import Texttable
  
  from aqt.exceptions import ArchiveConnectionError, ArchiveDownloadError, ArchiveListError, CliInputError, EmptyMetadata
  from aqt.helper import Settings, get_hash, getUrl, xml_to_modules
  
  
  class SimpleSpec(SemanticSimpleSpec):
      pass
  
      @staticmethod
      def usage() -> str:
          return (
              "See documentation at: "
              "https://python-semanticversion.readthedocs.io/en/latest/reference.html#semantic_version.SimpleSpec\n"
              "Examples:\n"
              '* "*": matches everything\n'
              '* "5": matches every version with major=5\n'
              '* "5.6": matches every version beginning with 5.6\n'
              '* "5.*.3": matches versions with major=5 and patch=3'
          )
  
  
  class Version(SemanticVersion):
      """Override semantic_version.Version class
      to accept Qt versions and tools versions
      If the version ends in `-preview`, the version is treated as a preview release.
      """
  
      def __init__(
          self,
          version_string=None,
          major=None,
          minor=None,
          patch=None,
          prerelease=None,
          build=None,
          partial=False,
      ):
          if version_string is None:
              super(Version, self).__init__(
                  version_string=None,
                  major=major,
                  minor=minor,
                  patch=patch,
                  prerelease=prerelease,
                  build=build,
                  partial=partial,
              )
              return
          # test qt versions
          match = re.match(r"^(\d+)\.(\d+)(\.(\d+)|-preview)$", version_string)
          if not match:
              # bad input
              raise ValueError("Invalid version string: '{}'".format(version_string))
          major, minor, end, patch = match.groups()
          is_preview = end == "-preview"
          super(Version, self).__init__(
              major=int(major),
              minor=int(minor),
              patch=int(patch) if patch else 0,
              prerelease=("preview",) if is_preview else None,
          )
  
      def __str__(self):
          if self.prerelease:
              return "{}.{}-preview".format(self.major, self.minor)
          return super(Version, self).__str__()
  
      @classmethod
      def permissive(cls, version_string: str):
          """Converts a version string with dots (5.X.Y, etc) into a semantic version.
          If the version omits either the patch or minor versions, they will be filled in with zeros,
          and the remaining version string becomes part of the prerelease component.
          If the version cannot be converted to a Version, a ValueError is raised.
  
          This is intended to be used on Version tags in an Updates.xml file.
  
          '1.33.1-202102101246' => Version('1.33.1-202102101246')
          '1.33-202102101246' => Version('1.33.0-202102101246')    # tools_conan
          '2020-05-19-1' => Version('2020.0.0-05-19-1')            # tools_vcredist
          """
  
          match = re.match(r"^(\d+)(\.(\d+)(\.(\d+))?)?(-(.+))?$", version_string)
          if not match:
              raise ValueError("Invalid version string: '{}'".format(version_string))
          major, dot_minor, minor, dot_patch, patch, hyphen_build, build = match.groups()
          return cls(
              major=int(major),
              minor=int(minor) if minor else 0,
              patch=int(patch) if patch else 0,
              build=(build,) if build else None,
          )
  
  
  class Versions:
      def __init__(
          self,
          versions: Union[None, Version, Iterable[Tuple[int, Iterable[Version]]]],
      ):
          if versions is None:
              self.versions = list()
          elif isinstance(versions, Version):
              self.versions = [[versions]]
          else:
              self.versions: List[List[Version]] = [list(versions_iterator) for _, versions_iterator in versions]
  
      def __str__(self) -> str:
          return str(self.versions)
  
      def __format__(self, format_spec) -> str:
          if format_spec == "":
              return "\n".join(" ".join(str(version) for version in minor_list) for minor_list in self.versions)
          elif format_spec == "s":
              return str(self.versions)
          else:
              raise TypeError("Unsupported format.")
  
      def __bool__(self):
          return len(self.versions) > 0 and len(self.versions[0]) > 0
  
      def latest(self) -> Optional[Version]:
          if not self:
              return None
          return self.versions[-1][-1]
  
      def __iter__(self) -> Generator[List[Version], None, None]:
          for item in self.versions:
              yield item
  
      def flattened(self) -> List[Version]:
          """Return a flattened list of all versions"""
          return [version for row in self for version in row]
  
  
  def get_semantic_version(qt_ver: str, is_preview: bool) -> Optional[Version]:
      """Converts a Qt version string (596, 512, 5132, etc) into a semantic version.
      This makes a lot of assumptions based on established patterns:
      If is_preview is True, the number is interpreted as ver[0].ver[1:], with no patch.
      If the version is 3 digits, then major, minor, and patch each get 1 digit.
      If the version is 4 or more digits, then major gets 1 digit, minor gets 2 digits
      and patch gets all the rest.
      As of May 2021, the version strings at https://download.qt.io/online/qtsdkrepository
      conform to this pattern; they are not guaranteed to do so in the future.
      """
      if not qt_ver or any(not ch.isdigit() for ch in qt_ver):
          return None
      if is_preview:
          return Version(
              major=int(qt_ver[:1]),
              minor=int(qt_ver[1:]),
              patch=0,
              prerelease=("preview",),
          )
      elif len(qt_ver) >= 4:
          return Version(major=int(qt_ver[:1]), minor=int(qt_ver[1:3]), patch=int(qt_ver[3:]))
      elif len(qt_ver) == 3:
          return Version(major=int(qt_ver[:1]), minor=int(qt_ver[1:2]), patch=int(qt_ver[2:]))
      elif len(qt_ver) == 2:
          return Version(major=int(qt_ver[:1]), minor=int(qt_ver[1:2]), patch=0)
      raise ValueError("Invalid version string '{}'".format(qt_ver))
  
  
  class ArchiveId:
      CATEGORIES = ("tools", "qt")
      HOSTS = ("windows", "mac", "linux")
      TARGETS_FOR_HOST = {
          "windows": ["android", "desktop", "winrt"],
          "mac": ["android", "desktop", "ios"],
          "linux": ["android", "desktop"],
      }
      ALL_EXTENSIONS = (
          "wasm",
          "src_doc_examples",
          "preview",
          "wasm_preview",
          "x86_64",
          "x86",
          "armv7",
          "arm64_v8a",
      )
      EXTENSIONS_REQUIRED_ANDROID_QT6 = "x86_64", "x86", "armv7", "arm64_v8a"
  
      def __init__(self, category: str, host: str, target: str, extension: str = ""):
          if category not in ArchiveId.CATEGORIES:
              raise ValueError("Category '{}' is invalid".format(category))
          if host not in ArchiveId.HOSTS:
              raise ValueError("Host '{}' is invalid".format(host))
          if target not in ArchiveId.TARGETS_FOR_HOST[host]:
              raise ValueError("Target '{}' is invalid".format(target))
          if extension and extension not in ArchiveId.ALL_EXTENSIONS:
              raise ValueError("Extension '{}' is invalid".format(extension))
          self.category: str = category
          self.host: str = host
          self.target: str = target
          self.extension: str = extension
  
      def is_preview(self) -> bool:
          return "preview" in self.extension if self.extension else False
  
      def is_qt(self) -> bool:
          return self.category == "qt"
  
      def is_tools(self) -> bool:
          return self.category == "tools"
  
      def is_no_arch(self) -> bool:
          """Returns True if there should be no arch attached to the module names"""
          return self.extension in ("src_doc_examples",)
  
      def to_url(self) -> str:
          return "online/qtsdkrepository/{os}{arch}/{target}/".format(
              os=self.host,
              arch="_x86" if self.host == "windows" else "_x64",
              target=self.target,
          )
  
      def to_folder(self, qt_version_no_dots: str) -> str:
          return "{category}{major}_{ver}{ext}".format(
              category=self.category,
              major=qt_version_no_dots[0],
              ver=qt_version_no_dots,
              ext="_" + self.extension if self.extension else "",
          )
  
      def __str__(self) -> str:
          return "{cat}/{host}/{target}{ext}".format(
              cat=self.category,
              host=self.host,
              target=self.target,
              ext="" if not self.extension else "/" + self.extension,
          )
  
  
  class ToolData:
      """A data class hold tool details."""
  
      head = [
          "Tool Variant Name",
          "Version",
          "Release Date",
          "Display Name",
          "Description",
      ]
  
      short_head = [
          "Tool Variant Name",
          "Version",
          "Release Date",
      ]
  
      def __init__(self, tool_data: Dict[str, Dict[str, str]]):
          self.tool_data: Dict[str, Dict[str, str]] = tool_data
          for key in tool_data.keys():
              self.tool_data[key]["Description"] = tool_data[key]["Description"].replace("<br>", "\n")
  
      def __format__(self, format_spec: str) -> str:
          short = False
          if format_spec == "{:s}":
              return str(self)
          if format_spec == "":
              max_width: int = 0
          elif format_spec == "{:T}":
              short = True
              max_width = 0
          else:
              match = re.match(r"\{?:?(\d+)t\}?", format_spec)
              if match:
                  g = match.groups()
                  max_width = int(g[0])
              else:
                  raise ValueError("Wrong format {}".format(format_spec))
          table = Texttable(max_width=max_width)
          table.set_deco(Texttable.HEADER)
          if short:
              table.header(self.short_head)
              table.add_rows(self._short_rows(), header=False)
          else:
              table.header(self.head)
              table.add_rows(self._rows(), header=False)
          return table.draw()
  
      def __bool__(self):
          return bool(self.tool_data)
  
      def _rows(self):
          keys = ("Version", "ReleaseDate", "DisplayName", "Description")
          return [[name, *[content[key] for key in keys]] for name, content in self.tool_data.items()]
  
      def _short_rows(self):
          keys = ("Version", "ReleaseDate")
          return [[name, *[content[key] for key in keys]] for name, content in self.tool_data.items()]
  
  
  class QtRepoProperty:
      """
      Describes properties of the Qt repository at https://download.qt.io/online/qtsdkrepository.
      Intended to help decouple the logic of aqt from specific properties of the Qt repository.
      """
  
      @staticmethod
      def extension_for_arch(architecture: str, is_version_ge_6: bool) -> str:
          if architecture == "wasm_32":
              return "wasm"
          elif architecture.startswith("android_") and is_version_ge_6:
              ext = architecture[len("android_") :]
              if ext in ArchiveId.EXTENSIONS_REQUIRED_ANDROID_QT6:
                  return ext
          return ""
  
      @staticmethod
      def possible_extensions_for_arch(arch: str) -> List[str]:
          """Assumes no knowledge of the Qt version"""
  
          # ext_ge_6: the extension if the version is greater than or equal to 6.0.0
          # ext_lt_6: the extension if the version is less than 6.0.0
          ext_lt_6, ext_ge_6 = [QtRepoProperty.extension_for_arch(arch, is_ge_6) for is_ge_6 in (False, True)]
          if ext_lt_6 == ext_ge_6:
              return [ext_lt_6]
          return [ext_lt_6, ext_ge_6]
  
  
  class MetadataFactory:
      """Retrieve metadata of Qt variations, versions, and descriptions from Qt site."""
  
      def __init__(
          self,
          archive_id: ArchiveId,
          *,
          spec: Optional[SimpleSpec] = None,
          is_latest_version: bool = False,
          modules_query: Optional[Tuple[str, str]] = None,
          extensions_ver: Optional[str] = None,
          architectures_ver: Optional[str] = None,
          archives_query: Optional[List[str]] = None,
          src_doc_examples_query: Optional[Tuple[str, Version, bool]] = None,
          tool_name: Optional[str] = None,
          is_long_listing: bool = False,
      ):
          """
          Construct MetadataFactory.
  
          :param spec:                When set, the MetadataFactory will filter out all versions of
                                      Qt that don't fit this SimpleSpec.
          :param is_latest_version:   When True, the MetadataFactory will find all versions of Qt
                                      matching filters, and only print the most recent version
          :param modules_query:       [Version of Qt, architecture] for which to list modules
          :param extensions_ver:      Version of Qt for which to list extensions
          :param architectures_ver:   Version of Qt for which to list architectures
          :param archives_query:      [Qt_Version, architecture, *module_names]: used to print list of archives
          :param tool_name:           Name of a tool, without architecture, ie "tools_qtcreator" or "tools_ifw"
          :param is_long_listing:     If true, long listing is used for tools output
          """
          self.logger = getLogger("aqt.metadata")
          self.archive_id = archive_id
          self.spec = spec
  
          if archive_id.is_tools():
              if tool_name:
                  if not tool_name.startswith("tools_"):
                      tool_name = "tools_" + tool_name
                  if is_long_listing:
                      self.request_type = "tool long listing"
                      self._action = lambda: self.fetch_tool_long_listing(tool_name)
                  else:
                      self.request_type = "tool variant names"
                      self._action = lambda: self.fetch_tool_modules(tool_name)
              else:
                  self.request_type = "tools"
                  self._action = self.fetch_tools
          elif is_latest_version:
              self.request_type = "latest version"
              self._action = lambda: Versions(self.fetch_latest_version())
          elif modules_query:
              self.request_type = "modules"
              version, arch = modules_query
              self._action = lambda: self.fetch_modules(self._to_version(version), arch)
          elif extensions_ver:
              self.request_type = "extensions"
              self._action = lambda: self.fetch_extensions(self._to_version(extensions_ver))
          elif architectures_ver:
              self.request_type = "architectures"
              self._action = lambda: self.fetch_arches(self._to_version(architectures_ver))
          elif archives_query:
              if len(archives_query) < 2:
                  raise CliInputError("The '--archives' flag requires a 'QT_VERSION' and an 'ARCHITECTURE' parameter.")
              self.request_type = "archives for modules" if len(archives_query) > 2 else "archives for qt"
              version, arch, modules = archives_query[0], archives_query[1], archives_query[2:]
              self._action = lambda: self.fetch_archives(self._to_version(version), arch, modules)
          elif src_doc_examples_query:
              cmd_type, version, is_modules_query = src_doc_examples_query
              if is_modules_query:
                  self.request_type = f"modules for {cmd_type}"
                  self._action = lambda: self.fetch_modules_sde(cmd_type, version)
              else:
                  self.request_type = f"archives for {cmd_type}"
                  self._action = lambda: self.fetch_archives_sde(cmd_type, version)
          else:
              self.request_type = "versions"
              self._action = self.fetch_versions
  
      def getList(self) -> Union[List[str], Versions, ToolData]:
          return self._action()
  
      def fetch_arches(self, version: Version) -> List[str]:
          self.validate_extension(version)
          if self.archive_id.extension == "src_doc_examples":
              return []
          qt_ver_str = self._get_qt_version_str(version)
          modules = self._fetch_module_metadata(self.archive_id.to_folder(qt_ver_str))
  
          arches = []
          for name in modules.keys():
              ver, arch = name.split(".")[-2:]
              if ver == qt_ver_str:
                  arches.append(arch)
  
          return arches
  
      def fetch_extensions(self, version: Version) -> List[str]:
          versions_extensions = MetadataFactory.get_versions_extensions(
              self.fetch_http(self.archive_id.to_url(), False), self.archive_id.category
          )
          filtered = filter(
              lambda ver_ext: ver_ext[0] == version and ver_ext[1],
              versions_extensions,
          )
          return list(map(lambda ver_ext: ver_ext[1], filtered))
  
      def fetch_versions(self) -> Versions:
          def filter_by(ver_ext: Tuple[Optional[Version], str]) -> bool:
              version, extension = ver_ext
              return version and (self.spec is None or version in self.spec) and (self.archive_id.extension == extension)
  
          def get_version(ver_ext: Tuple[Version, str]):
              return ver_ext[0]
  
          versions_extensions = MetadataFactory.get_versions_extensions(
              self.fetch_http(self.archive_id.to_url(), False), self.archive_id.category
          )
          versions = sorted(filter(None, map(get_version, filter(filter_by, versions_extensions))))
          iterables = itertools.groupby(versions, lambda version: version.minor)
          return Versions(iterables)
  
      def fetch_latest_version(self) -> Optional[Version]:
          return self.fetch_versions().latest()
  
      def fetch_tools(self) -> List[str]:
          html_doc = self.fetch_http(self.archive_id.to_url(), False)
          return list(MetadataFactory.iterate_folders(html_doc, "tools"))
  
      def fetch_tool_modules(self, tool_name: str) -> List[str]:
          tool_data = self._fetch_module_metadata(tool_name)
          return list(tool_data.keys())
  
      def fetch_tool_by_simple_spec(self, tool_name: str, simple_spec: SimpleSpec) -> Optional[Dict[str, str]]:
          # Get data for all the tool modules
          all_tools_data = self._fetch_module_metadata(tool_name)
          return self.choose_highest_version_in_spec(all_tools_data, simple_spec)
  
      def fetch_tool_long_listing(self, tool_name: str) -> ToolData:
          return ToolData(self._fetch_module_metadata(tool_name))
  
      def validate_extension(self, qt_ver: Version) -> None:
          """
          Checks extension, and raises CliInputError if invalid.
  
          Rules:
          1. On Qt6 for Android, an extension for processor architecture is required.
          2. On any platform other than Android, or on Qt5, an extension for
          processor architecture is forbidden.
          3. The "wasm" extension only works on desktop targets for Qt 5.13-5.15, or for 6.2+
          """
          if (
              self.archive_id.target == "android"
              and qt_ver.major == 6
              and self.archive_id.extension not in ArchiveId.EXTENSIONS_REQUIRED_ANDROID_QT6
          ):
              raise CliInputError(
                  "Qt 6 for Android requires one of the following extensions: "
                  f"{ArchiveId.EXTENSIONS_REQUIRED_ANDROID_QT6}. "
                  "Please add your extension using the `--extension` flag."
              )
          if self.archive_id.extension in ArchiveId.EXTENSIONS_REQUIRED_ANDROID_QT6 and (
              self.archive_id.target != "android" or qt_ver.major != 6
          ):
              raise CliInputError(f"The extension '{self.archive_id.extension}' is only valid for Qt 6 for Android")
          is_in_wasm_range = qt_ver in SimpleSpec(">=5.13,<6") or qt_ver in SimpleSpec(">=6.2.0")
          if "wasm" in self.archive_id.extension and (self.archive_id.target != "desktop" or not is_in_wasm_range):
              raise CliInputError(
                  f"The extension '{self.archive_id.extension}' is only available in Qt 5.13-5.15 and 6.2+ on desktop."
              )
  
      @staticmethod
      def choose_highest_version_in_spec(
          all_tools_data: Dict[str, Dict[str, str]], simple_spec: SimpleSpec
      ) -> Optional[Dict[str, str]]:
          # Get versions of all modules. Fail if version cannot be determined.
          try:
              tools_versions = [
                  (name, tool_data, Version.permissive(tool_data["Version"])) for name, tool_data in all_tools_data.items()
              ]
          except ValueError:
              return None
  
          # Remove items that don't conform to simple_spec
          tools_versions = filter(lambda tool_item: tool_item[2] in simple_spec, tools_versions)
  
          try:
              # Return the conforming item with the highest version.
              # If there are multiple items with the same version, the result will not be predictable.
              return max(tools_versions, key=operator.itemgetter(2))[1]
          except ValueError:
              # There were no tools that fit the simple_spec
              return None
  
      def _to_version(self, qt_ver: str) -> Version:
          """
          Turns a string in the form of `5.X.Y | latest` into a semantic version.
          If the string does not fit either of these forms, CliInputError will be raised.
          If qt_ver == latest, and no versions exist corresponding to the filters specified,
          then CliInputError will be raised.
          If qt_ver == latest, and an HTTP error occurs, requests.RequestException will be raised.
  
          :param qt_ver:  Either the literal string `latest`, or a semantic version
                          with each part separated with dots.
          """
          assert qt_ver
          if qt_ver == "latest":
              latest_version = self.fetch_latest_version()
              if not latest_version:
                  msg = "There is no latest version of Qt with the criteria '{}'".format(self.describe_filters())
                  raise CliInputError(msg)
              return latest_version
          try:
              version = Version(qt_ver)
          except ValueError as e:
              raise CliInputError(e) from e
          return version
  
      @staticmethod
      def fetch_http(rest_of_url: str, is_check_hash: bool = True) -> str:
          timeout = (Settings.connection_timeout, Settings.response_timeout)
          expected_hash = get_hash(rest_of_url, "sha256", timeout) if is_check_hash else None
          base_urls = Settings.baseurl, random.choice(Settings.fallbacks)
          for i, base_url in enumerate(base_urls):
              try:
                  url = posixpath.join(base_url, rest_of_url)
                  return getUrl(url=url, timeout=timeout, expected_hash=expected_hash)
  
              except (ArchiveDownloadError, ArchiveConnectionError) as e:
                  if i == len(base_urls) - 1:
                      raise e from e
                  else:
                      getLogger("aqt.metadata").debug(
                          f"Connection to '{base_url}' failed. Retrying with fallback '{base_urls[i + 1]}'."
                      )
  
      @staticmethod
      def iterate_folders(html_doc: str, filter_category: str = "") -> Generator[str, None, None]:
          def table_row_to_folder(tr: bs4.element.Tag) -> str:
              try:
                  return tr.find_all("td")[1].a.contents[0].rstrip("/")
              except (AttributeError, IndexError):
                  return ""
  
          soup: bs4.BeautifulSoup = bs4.BeautifulSoup(html_doc, "html.parser")
          for row in soup.body.table.find_all("tr"):
              content: str = table_row_to_folder(row)
              if not content or content == "Parent Directory":
                  continue
              if content.startswith(filter_category):
                  yield content
  
      @staticmethod
      def get_versions_extensions(html_doc: str, category: str) -> Iterator[Tuple[Optional[Version], str]]:
          def folder_to_version_extension(folder: str) -> Tuple[Optional[Version], str]:
              components = folder.split("_", maxsplit=2)
              ext = "" if len(components) < 3 else components[2]
              ver = "" if len(components) < 2 else components[1]
              return (
                  get_semantic_version(qt_ver=ver, is_preview="preview" in ext),
                  ext,
              )
  
          return map(
              folder_to_version_extension,
              MetadataFactory.iterate_folders(html_doc, category),
          )
  
      @staticmethod
      def _has_nonempty_downloads(element: Element) -> bool:
          """Returns True if the element has a nonempty '<DownloadableArchives/>' tag"""
          downloads = element.find("DownloadableArchives")
          update_file = element.find("UpdateFile")
          if downloads is None or update_file is None:
              return False
          uncompressed_size = int(update_file.attrib["UncompressedSize"])
          return downloads.text and uncompressed_size >= Settings.min_module_size
  
      def _get_qt_version_str(self, version: Version) -> str:
          """Returns a Qt version, without dots, that works in the Qt repo urls and Updates.xml files"""
          # NOTE: The url at `<base>/<host>/<target>/qt5_590/` does not exist; the real one is `qt5_59`
          patch = (
              ""
              if version.prerelease or self.archive_id.is_preview() or version in SimpleSpec("5.9.0")
              else str(version.patch)
          )
          return f"{version.major}{version.minor}{patch}"
  
      def _fetch_module_metadata(self, folder: str, predicate: Optional[Callable[[Element], bool]] = None):
          rest_of_url = posixpath.join(self.archive_id.to_url(), folder, "Updates.xml")
          xml = self.fetch_http(rest_of_url)
          return xml_to_modules(
              xml,
              predicate=predicate if predicate else MetadataFactory._has_nonempty_downloads,
          )
  
      def fetch_modules(self, version: Version, arch: str) -> List[str]:
          """Returns list of modules"""
          self.validate_extension(version)
          qt_ver_str = self._get_qt_version_str(version)
          # Example: re.compile(r"^(preview\.)?qt\.(qt5\.)?590\.(.+)$")
          pattern = re.compile(r"^(preview\.)?qt\.(qt" + str(version.major) + r"\.)?" + qt_ver_str + r"\.(.+)$")
          modules_meta = self._fetch_module_metadata(self.archive_id.to_folder(qt_ver_str))
  
          def to_module_arch(name: str) -> Tuple[Optional[str], Optional[str]]:
              _match = pattern.match(name)
              if not _match:
                  return None, None
              module_with_arch = _match.group(3)
              if self.archive_id.is_no_arch() or "." not in module_with_arch:
                  return module_with_arch, None
              module, arch = module_with_arch.rsplit(".", 1)
              if module.startswith("addons."):
                  module = module[len("addons.") :]
              return module, arch
  
          modules = set()
          for name in modules_meta.keys():
              module, _arch = to_module_arch(name)
              if _arch == arch:
                  modules.add(module)
          return sorted(modules)
  
      def fetch_modules_sde(self, cmd_type: str, version: Version) -> List[str]:
          """Returns list of modules for src/doc/examples"""
          assert (
              cmd_type in ("doc", "examples") and self.archive_id.target == "desktop"
          ), "Internal misuse of fetch_modules_sde"
          self.validate_extension(version)
          qt_ver_str = self._get_qt_version_str(version)
          modules_meta = self._fetch_module_metadata(self.archive_id.to_folder(qt_ver_str))
          # pattern: Match all names "qt.qt5.12345.doc.(\w+)
          pattern = re.compile(r"^qt\.(qt" + str(version.major) + r"\.)?" + qt_ver_str + r"\." + cmd_type + r"\.(.+)$")
  
          modules: List[str] = []
          for name in modules_meta:
              _match = pattern.match(name)
              if _match:
                  modules.append(_match.group(2))
          return modules
  
      def fetch_archives_sde(self, cmd_type: str, version: Version) -> List[str]:
          """Returns list of archives for src/doc/examples"""
          assert (
              cmd_type in ("src", "doc", "examples") and self.archive_id.target == "desktop"
          ), "Internal misuse of fetch_archives_sde"
          return self.fetch_archives(version, cmd_type, [])
  
      def fetch_archives(self, version: Version, arch: str, modules: List[str]) -> List[str]:
          qt_version_str = self._get_qt_version_str(version)
          nonempty = MetadataFactory._has_nonempty_downloads
  
          def all_modules(element: Element) -> bool:
              _module, _arch = element.find("Name").text.split(".")[-2:]
              return _arch == arch and _module != qt_version_str and nonempty(element)
  
          def specify_modules(element: Element) -> bool:
              _module, _arch = element.find("Name").text.split(".")[-2:]
              return _arch == arch and _module in modules and nonempty(element)
  
          def no_modules(element: Element) -> bool:
              name: Optional[str] = element.find("Name").text
              return name and name.endswith(f".{qt_version_str}.{arch}") and nonempty(element)
  
          predicate = no_modules if not modules else all_modules if "all" in modules else specify_modules
          try:
              mod_metadata = self._fetch_module_metadata(self.archive_id.to_folder(qt_version_str), predicate=predicate)
          except (AttributeError,) as e:
              raise ArchiveListError(f"Downloaded metadata is corrupted. {e}") from e
  
          # Did we find all requested modules?
          if modules and "all" not in modules:
              requested_set = set(modules)
              actual_set = set([_name.split(".")[-2] for _name in mod_metadata.keys()])
              not_found = sorted(requested_set.difference(actual_set))
              if not_found:
                  raise CliInputError(
                      f"The requested modules were not located: {not_found}", suggested_action=suggested_follow_up(self)
                  )
  
          csv_lists = [mod["DownloadableArchives"] for mod in mod_metadata.values()]
          return sorted(set([arc.split("-")[0] for csv in csv_lists for arc in csv.split(", ")]))
  
      def describe_filters(self) -> str:
          if self.spec is None:
              return str(self.archive_id)
          return "{} with spec {}".format(self.archive_id, self.spec)
  
  
  def suggested_follow_up(meta: MetadataFactory) -> List[str]:
      """Makes an informed guess at what the user got wrong, in the event of an error."""
      msg = []
      list_cmd = "list-tool" if meta.archive_id.is_tools() else "list-qt"
      base_cmd = "aqt {0} {1.host} {1.target}".format(list_cmd, meta.archive_id)
      versions_msg = f"Please use '{base_cmd}' to show versions of Qt available."
      arches_msg = f"Please use '{base_cmd} --arch <QT_VERSION>' to show architectures available."
      if meta.archive_id.extension:
          msg.append(f"Please use '{base_cmd} --extensions <QT_VERSION>' to list valid extensions.")
  
      if meta.archive_id.is_tools() and meta.request_type == "tool variant names":
          msg.append(f"Please use '{base_cmd}' to check what tools are available.")
      elif meta.spec is not None:
          msg.append(
              f"Please use '{base_cmd}' to check that versions of {meta.archive_id.category} "
              f"exist within the spec '{meta.spec}'."
          )
      elif meta.request_type in ("architectures", "modules", "extensions"):
          msg.append(f"Please use '{base_cmd}' to show versions of Qt available.")
          if meta.request_type == "modules":
              msg.append(f"Please use '{base_cmd} --arch <QT_VERSION>' to list valid architectures.")
      elif meta.request_type == "archives for modules":
          msg.extend([versions_msg, arches_msg, f"Please use '{base_cmd} --modules <QT_VERSION>' to show modules available."])
      elif meta.request_type == "archives for qt":
          msg.extend([versions_msg, arches_msg])
  
      return msg
  
  
  def show_list(meta: MetadataFactory):
      try:
          output = meta.getList()
          if not output:
              raise EmptyMetadata(
                  f"No {meta.request_type} available for this request.", suggested_action=suggested_follow_up(meta)
              )
          if isinstance(output, Versions):
              print(format(output))
          elif isinstance(output, ToolData):
              width: int = shutil.get_terminal_size((0, 40)).columns
              if width == 0:  # notty ?
                  print(format(output, "{:0t}"))
              elif width < 95:  # narrow terminal
                  print(format(output, "{:T}"))
              else:
                  print("{0:{1}t}".format(output, width))
          elif meta.archive_id.is_tools():
              print(*output, sep="\n")
          else:
              print(*output, sep=" ")
      except (ArchiveDownloadError, ArchiveConnectionError) as e:
          e.suggested_action = suggested_follow_up(meta)
          raise e from e
  
files.new_path: aqt/metadata.py
files.old_contents: #!/usr/bin/env python
  #
  # Copyright (C) 2021 David Dalcino
  #
  # Permission is hereby granted, free of charge, to any person obtaining a copy of
  # this software and associated documentation files (the "Software"), to deal in
  # the Software without restriction, including without limitation the rights to
  # use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
  # the Software, and to permit persons to whom the Software is furnished to do so,
  # subject to the following conditions:
  #
  # The above copyright notice and this permission notice shall be included in all
  # copies or substantial portions of the Software.
  #
  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
  # FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
  # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
  # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  import binascii
  import itertools
  import operator
  import posixpath
  import re
  import secrets as random
  import shutil
  from logging import getLogger
  from typing import Callable, Dict, Generator, Iterable, Iterator, List, Optional, Tuple, Union
  from xml.etree.ElementTree import Element
  
  import bs4
  from semantic_version import SimpleSpec as SemanticSimpleSpec
  from semantic_version import Version as SemanticVersion
  from texttable import Texttable
  
  from aqt.exceptions import ArchiveConnectionError, ArchiveDownloadError, ArchiveListError, CliInputError, EmptyMetadata
  from aqt.helper import Settings, get_hash, getUrl, xml_to_modules
  
  
  class SimpleSpec(SemanticSimpleSpec):
      pass
  
      @staticmethod
      def usage() -> str:
          return (
              "See documentation at: "
              "https://python-semanticversion.readthedocs.io/en/latest/reference.html#semantic_version.SimpleSpec\n"
              "Examples:\n"
              '* "*": matches everything\n'
              '* "5": matches every version with major=5\n'
              '* "5.6": matches every version beginning with 5.6\n'
              '* "5.*.3": matches versions with major=5 and patch=3'
          )
  
  
  class Version(SemanticVersion):
      """Override semantic_version.Version class
      to accept Qt versions and tools versions
      If the version ends in `-preview`, the version is treated as a preview release.
      """
  
      def __init__(
          self,
          version_string=None,
          major=None,
          minor=None,
          patch=None,
          prerelease=None,
          build=None,
          partial=False,
      ):
          if version_string is None:
              super(Version, self).__init__(
                  version_string=None,
                  major=major,
                  minor=minor,
                  patch=patch,
                  prerelease=prerelease,
                  build=build,
                  partial=partial,
              )
              return
          # test qt versions
          match = re.match(r"^(\d+)\.(\d+)(\.(\d+)|-preview)$", version_string)
          if not match:
              # bad input
              raise ValueError("Invalid version string: '{}'".format(version_string))
          major, minor, end, patch = match.groups()
          is_preview = end == "-preview"
          super(Version, self).__init__(
              major=int(major),
              minor=int(minor),
              patch=int(patch) if patch else 0,
              prerelease=("preview",) if is_preview else None,
          )
  
      def __str__(self):
          if self.prerelease:
              return "{}.{}-preview".format(self.major, self.minor)
          return super(Version, self).__str__()
  
      @classmethod
      def permissive(cls, version_string: str):
          """Converts a version string with dots (5.X.Y, etc) into a semantic version.
          If the version omits either the patch or minor versions, they will be filled in with zeros,
          and the remaining version string becomes part of the prerelease component.
          If the version cannot be converted to a Version, a ValueError is raised.
  
          This is intended to be used on Version tags in an Updates.xml file.
  
          '1.33.1-202102101246' => Version('1.33.1-202102101246')
          '1.33-202102101246' => Version('1.33.0-202102101246')    # tools_conan
          '2020-05-19-1' => Version('2020.0.0-05-19-1')            # tools_vcredist
          """
  
          match = re.match(r"^(\d+)(\.(\d+)(\.(\d+))?)?(-(.+))?$", version_string)
          if not match:
              raise ValueError("Invalid version string: '{}'".format(version_string))
          major, dot_minor, minor, dot_patch, patch, hyphen_build, build = match.groups()
          return cls(
              major=int(major),
              minor=int(minor) if minor else 0,
              patch=int(patch) if patch else 0,
              build=(build,) if build else None,
          )
  
  
  class Versions:
      def __init__(
          self,
          versions: Union[None, Version, Iterable[Tuple[int, Iterable[Version]]]],
      ):
          if versions is None:
              self.versions = list()
          elif isinstance(versions, Version):
              self.versions = [[versions]]
          else:
              self.versions: List[List[Version]] = [list(versions_iterator) for _, versions_iterator in versions]
  
      def __str__(self) -> str:
          return str(self.versions)
  
      def __format__(self, format_spec) -> str:
          if format_spec == "":
              return "\n".join(" ".join(str(version) for version in minor_list) for minor_list in self.versions)
          elif format_spec == "s":
              return str(self.versions)
          else:
              raise TypeError("Unsupported format.")
  
      def __bool__(self):
          return len(self.versions) > 0 and len(self.versions[0]) > 0
  
      def latest(self) -> Optional[Version]:
          if not self:
              return None
          return self.versions[-1][-1]
  
      def __iter__(self) -> Generator[List[Version], None, None]:
          for item in self.versions:
              yield item
  
      def flattened(self) -> List[Version]:
          """Return a flattened list of all versions"""
          return [version for row in self for version in row]
  
  
  def get_semantic_version(qt_ver: str, is_preview: bool) -> Optional[Version]:
      """Converts a Qt version string (596, 512, 5132, etc) into a semantic version.
      This makes a lot of assumptions based on established patterns:
      If is_preview is True, the number is interpreted as ver[0].ver[1:], with no patch.
      If the version is 3 digits, then major, minor, and patch each get 1 digit.
      If the version is 4 or more digits, then major gets 1 digit, minor gets 2 digits
      and patch gets all the rest.
      As of May 2021, the version strings at https://download.qt.io/online/qtsdkrepository
      conform to this pattern; they are not guaranteed to do so in the future.
      """
      if not qt_ver or any(not ch.isdigit() for ch in qt_ver):
          return None
      if is_preview:
          return Version(
              major=int(qt_ver[:1]),
              minor=int(qt_ver[1:]),
              patch=0,
              prerelease=("preview",),
          )
      elif len(qt_ver) >= 4:
          return Version(major=int(qt_ver[:1]), minor=int(qt_ver[1:3]), patch=int(qt_ver[3:]))
      elif len(qt_ver) == 3:
          return Version(major=int(qt_ver[:1]), minor=int(qt_ver[1:2]), patch=int(qt_ver[2:]))
      elif len(qt_ver) == 2:
          return Version(major=int(qt_ver[:1]), minor=int(qt_ver[1:2]), patch=0)
      raise ValueError("Invalid version string '{}'".format(qt_ver))
  
  
  class ArchiveId:
      CATEGORIES = ("tools", "qt")
      HOSTS = ("windows", "mac", "linux")
      TARGETS_FOR_HOST = {
          "windows": ["android", "desktop", "winrt"],
          "mac": ["android", "desktop", "ios"],
          "linux": ["android", "desktop"],
      }
      ALL_EXTENSIONS = (
          "wasm",
          "src_doc_examples",
          "preview",
          "wasm_preview",
          "x86_64",
          "x86",
          "armv7",
          "arm64_v8a",
      )
      EXTENSIONS_REQUIRED_ANDROID_QT6 = "x86_64", "x86", "armv7", "arm64_v8a"
  
      def __init__(self, category: str, host: str, target: str, extension: str = ""):
          if category not in ArchiveId.CATEGORIES:
              raise ValueError("Category '{}' is invalid".format(category))
          if host not in ArchiveId.HOSTS:
              raise ValueError("Host '{}' is invalid".format(host))
          if target not in ArchiveId.TARGETS_FOR_HOST[host]:
              raise ValueError("Target '{}' is invalid".format(target))
          if extension and extension not in ArchiveId.ALL_EXTENSIONS:
              raise ValueError("Extension '{}' is invalid".format(extension))
          self.category: str = category
          self.host: str = host
          self.target: str = target
          self.extension: str = extension
  
      def is_preview(self) -> bool:
          return "preview" in self.extension if self.extension else False
  
      def is_qt(self) -> bool:
          return self.category == "qt"
  
      def is_tools(self) -> bool:
          return self.category == "tools"
  
      def is_no_arch(self) -> bool:
          """Returns True if there should be no arch attached to the module names"""
          return self.extension in ("src_doc_examples",)
  
      def to_url(self) -> str:
          return "online/qtsdkrepository/{os}{arch}/{target}/".format(
              os=self.host,
              arch="_x86" if self.host == "windows" else "_x64",
              target=self.target,
          )
  
      def to_folder(self, qt_version_no_dots: str) -> str:
          return "{category}{major}_{ver}{ext}".format(
              category=self.category,
              major=qt_version_no_dots[0],
              ver=qt_version_no_dots,
              ext="_" + self.extension if self.extension else "",
          )
  
      def __str__(self) -> str:
          return "{cat}/{host}/{target}{ext}".format(
              cat=self.category,
              host=self.host,
              target=self.target,
              ext="" if not self.extension else "/" + self.extension,
          )
  
  
  class ToolData:
      """A data class hold tool details."""
  
      head = [
          "Tool Variant Name",
          "Version",
          "Release Date",
          "Display Name",
          "Description",
      ]
  
      short_head = [
          "Tool Variant Name",
          "Version",
          "Release Date",
      ]
  
      def __init__(self, tool_data: Dict[str, Dict[str, str]]):
          self.tool_data: Dict[str, Dict[str, str]] = tool_data
          for key in tool_data.keys():
              self.tool_data[key]["Description"] = tool_data[key]["Description"].replace("<br>", "\n")
  
      def __format__(self, format_spec: str) -> str:
          short = False
          if format_spec == "{:s}":
              return str(self)
          if format_spec == "":
              max_width: int = 0
          elif format_spec == "{:T}":
              short = True
              max_width = 0
          else:
              match = re.match(r"\{?:?(\d+)t\}?", format_spec)
              if match:
                  g = match.groups()
                  max_width = int(g[0])
              else:
                  raise ValueError("Wrong format {}".format(format_spec))
          table = Texttable(max_width=max_width)
          table.set_deco(Texttable.HEADER)
          if short:
              table.header(self.short_head)
              table.add_rows(self._short_rows(), header=False)
          else:
              table.header(self.head)
              table.add_rows(self._rows(), header=False)
          return table.draw()
  
      def __bool__(self):
          return bool(self.tool_data)
  
      def _rows(self):
          keys = ("Version", "ReleaseDate", "DisplayName", "Description")
          return [[name, *[content[key] for key in keys]] for name, content in self.tool_data.items()]
  
      def _short_rows(self):
          keys = ("Version", "ReleaseDate")
          return [[name, *[content[key] for key in keys]] for name, content in self.tool_data.items()]
  
  
  class QtRepoProperty:
      """
      Describes properties of the Qt repository at https://download.qt.io/online/qtsdkrepository.
      Intended to help decouple the logic of aqt from specific properties of the Qt repository.
      """
  
      @staticmethod
      def extension_for_arch(architecture: str, is_version_ge_6: bool) -> str:
          if architecture == "wasm_32":
              return "wasm"
          elif architecture.startswith("android_") and is_version_ge_6:
              ext = architecture[len("android_") :]
              if ext in ArchiveId.EXTENSIONS_REQUIRED_ANDROID_QT6:
                  return ext
          return ""
  
      @staticmethod
      def possible_extensions_for_arch(arch: str) -> List[str]:
          """Assumes no knowledge of the Qt version"""
  
          # ext_ge_6: the extension if the version is greater than or equal to 6.0.0
          # ext_lt_6: the extension if the version is less than 6.0.0
          ext_lt_6, ext_ge_6 = [QtRepoProperty.extension_for_arch(arch, is_ge_6) for is_ge_6 in (False, True)]
          if ext_lt_6 == ext_ge_6:
              return [ext_lt_6]
          return [ext_lt_6, ext_ge_6]
  
  
  class MetadataFactory:
      """Retrieve metadata of Qt variations, versions, and descriptions from Qt site."""
  
      def __init__(
          self,
          archive_id: ArchiveId,
          *,
          spec: Optional[SimpleSpec] = None,
          is_latest_version: bool = False,
          modules_query: Optional[Tuple[str, str]] = None,
          extensions_ver: Optional[str] = None,
          architectures_ver: Optional[str] = None,
          archives_query: Optional[List[str]] = None,
          src_doc_examples_query: Optional[Tuple[str, Version, bool]] = None,
          tool_name: Optional[str] = None,
          is_long_listing: bool = False,
      ):
          """
          Construct MetadataFactory.
  
          :param spec:                When set, the MetadataFactory will filter out all versions of
                                      Qt that don't fit this SimpleSpec.
          :param is_latest_version:   When True, the MetadataFactory will find all versions of Qt
                                      matching filters, and only print the most recent version
          :param modules_query:       [Version of Qt, architecture] for which to list modules
          :param extensions_ver:      Version of Qt for which to list extensions
          :param architectures_ver:   Version of Qt for which to list architectures
          :param archives_query:      [Qt_Version, architecture, *module_names]: used to print list of archives
          :param tool_name:           Name of a tool, without architecture, ie "tools_qtcreator" or "tools_ifw"
          :param is_long_listing:     If true, long listing is used for tools output
          """
          self.logger = getLogger("aqt.metadata")
          self.archive_id = archive_id
          self.spec = spec
  
          if archive_id.is_tools():
              if tool_name:
                  if not tool_name.startswith("tools_"):
                      tool_name = "tools_" + tool_name
                  if is_long_listing:
                      self.request_type = "tool long listing"
                      self._action = lambda: self.fetch_tool_long_listing(tool_name)
                  else:
                      self.request_type = "tool variant names"
                      self._action = lambda: self.fetch_tool_modules(tool_name)
              else:
                  self.request_type = "tools"
                  self._action = self.fetch_tools
          elif is_latest_version:
              self.request_type = "latest version"
              self._action = lambda: Versions(self.fetch_latest_version())
          elif modules_query:
              self.request_type = "modules"
              version, arch = modules_query
              self._action = lambda: self.fetch_modules(self._to_version(version), arch)
          elif extensions_ver:
              self.request_type = "extensions"
              self._action = lambda: self.fetch_extensions(self._to_version(extensions_ver))
          elif architectures_ver:
              self.request_type = "architectures"
              self._action = lambda: self.fetch_arches(self._to_version(architectures_ver))
          elif archives_query:
              if len(archives_query) < 2:
                  raise CliInputError("The '--archives' flag requires a 'QT_VERSION' and an 'ARCHITECTURE' parameter.")
              self.request_type = "archives for modules" if len(archives_query) > 2 else "archives for qt"
              version, arch, modules = archives_query[0], archives_query[1], archives_query[2:]
              self._action = lambda: self.fetch_archives(self._to_version(version), arch, modules)
          elif src_doc_examples_query:
              cmd_type, version, is_modules_query = src_doc_examples_query
              if is_modules_query:
                  self.request_type = f"modules for {cmd_type}"
                  self._action = lambda: self.fetch_modules_sde(cmd_type, version)
              else:
                  self.request_type = f"archives for {cmd_type}"
                  self._action = lambda: self.fetch_archives_sde(cmd_type, version)
          else:
              self.request_type = "versions"
              self._action = self.fetch_versions
  
      def getList(self) -> Union[List[str], Versions, ToolData]:
          return self._action()
  
      def fetch_arches(self, version: Version) -> List[str]:
          self.validate_extension(version)
          if self.archive_id.extension == "src_doc_examples":
              return []
          qt_ver_str = self._get_qt_version_str(version)
          modules = self._fetch_module_metadata(self.archive_id.to_folder(qt_ver_str))
  
          arches = []
          for name in modules.keys():
              ver, arch = name.split(".")[-2:]
              if ver == qt_ver_str:
                  arches.append(arch)
  
          return arches
  
      def fetch_extensions(self, version: Version) -> List[str]:
          versions_extensions = MetadataFactory.get_versions_extensions(
              self.fetch_http(self.archive_id.to_url(), False), self.archive_id.category
          )
          filtered = filter(
              lambda ver_ext: ver_ext[0] == version and ver_ext[1],
              versions_extensions,
          )
          return list(map(lambda ver_ext: ver_ext[1], filtered))
  
      def fetch_versions(self) -> Versions:
          def filter_by(ver_ext: Tuple[Optional[Version], str]) -> bool:
              version, extension = ver_ext
              return version and (self.spec is None or version in self.spec) and (self.archive_id.extension == extension)
  
          def get_version(ver_ext: Tuple[Version, str]):
              return ver_ext[0]
  
          versions_extensions = MetadataFactory.get_versions_extensions(
              self.fetch_http(self.archive_id.to_url(), False), self.archive_id.category
          )
          versions = sorted(filter(None, map(get_version, filter(filter_by, versions_extensions))))
          iterables = itertools.groupby(versions, lambda version: version.minor)
          return Versions(iterables)
  
      def fetch_latest_version(self) -> Optional[Version]:
          return self.fetch_versions().latest()
  
      def fetch_tools(self) -> List[str]:
          html_doc = self.fetch_http(self.archive_id.to_url(), False)
          return list(MetadataFactory.iterate_folders(html_doc, "tools"))
  
      def fetch_tool_modules(self, tool_name: str) -> List[str]:
          tool_data = self._fetch_module_metadata(tool_name)
          return list(tool_data.keys())
  
      def fetch_tool_by_simple_spec(self, tool_name: str, simple_spec: SimpleSpec) -> Optional[Dict[str, str]]:
          # Get data for all the tool modules
          all_tools_data = self._fetch_module_metadata(tool_name)
          return self.choose_highest_version_in_spec(all_tools_data, simple_spec)
  
      def fetch_tool_long_listing(self, tool_name: str) -> ToolData:
          return ToolData(self._fetch_module_metadata(tool_name))
  
      def validate_extension(self, qt_ver: Version) -> None:
          """
          Checks extension, and raises CliInputError if invalid.
  
          Rules:
          1. On Qt6 for Android, an extension for processor architecture is required.
          2. On any platform other than Android, or on Qt5, an extension for
          processor architecture is forbidden.
          3. The "wasm" extension only works on desktop targets for Qt 5.13-5.15, or for 6.2+
          """
          if (
              self.archive_id.target == "android"
              and qt_ver.major == 6
              and self.archive_id.extension not in ArchiveId.EXTENSIONS_REQUIRED_ANDROID_QT6
          ):
              raise CliInputError(
                  "Qt 6 for Android requires one of the following extensions: "
                  f"{ArchiveId.EXTENSIONS_REQUIRED_ANDROID_QT6}. "
                  "Please add your extension using the `--extension` flag."
              )
          if self.archive_id.extension in ArchiveId.EXTENSIONS_REQUIRED_ANDROID_QT6 and (
              self.archive_id.target != "android" or qt_ver.major != 6
          ):
              raise CliInputError(f"The extension '{self.archive_id.extension}' is only valid for Qt 6 for Android")
          is_in_wasm_range = qt_ver in SimpleSpec(">=5.13,<6") or qt_ver in SimpleSpec(">=6.2.0")
          if "wasm" in self.archive_id.extension and (self.archive_id.target != "desktop" or not is_in_wasm_range):
              raise CliInputError(
                  f"The extension '{self.archive_id.extension}' is only available in Qt 5.13-5.15 and 6.2+ on desktop."
              )
  
      @staticmethod
      def choose_highest_version_in_spec(
          all_tools_data: Dict[str, Dict[str, str]], simple_spec: SimpleSpec
      ) -> Optional[Dict[str, str]]:
          # Get versions of all modules. Fail if version cannot be determined.
          try:
              tools_versions = [
                  (name, tool_data, Version.permissive(tool_data["Version"])) for name, tool_data in all_tools_data.items()
              ]
          except ValueError:
              return None
  
          # Remove items that don't conform to simple_spec
          tools_versions = filter(lambda tool_item: tool_item[2] in simple_spec, tools_versions)
  
          try:
              # Return the conforming item with the highest version.
              # If there are multiple items with the same version, the result will not be predictable.
              return max(tools_versions, key=operator.itemgetter(2))[1]
          except ValueError:
              # There were no tools that fit the simple_spec
              return None
  
      def _to_version(self, qt_ver: str) -> Version:
          """
          Turns a string in the form of `5.X.Y | latest` into a semantic version.
          If the string does not fit either of these forms, CliInputError will be raised.
          If qt_ver == latest, and no versions exist corresponding to the filters specified,
          then CliInputError will be raised.
          If qt_ver == latest, and an HTTP error occurs, requests.RequestException will be raised.
  
          :param qt_ver:  Either the literal string `latest`, or a semantic version
                          with each part separated with dots.
          """
          assert qt_ver
          if qt_ver == "latest":
              latest_version = self.fetch_latest_version()
              if not latest_version:
                  msg = "There is no latest version of Qt with the criteria '{}'".format(self.describe_filters())
                  raise CliInputError(msg)
              return latest_version
          try:
              version = Version(qt_ver)
          except ValueError as e:
              raise CliInputError(e) from e
          return version
  
      @staticmethod
      def fetch_http(rest_of_url: str, is_check_hash: bool = True) -> str:
          timeout = (Settings.connection_timeout, Settings.response_timeout)
          expected_hash = binascii.unhexlify(get_hash(rest_of_url, "sha256", timeout)) if is_check_hash else None
          base_urls = Settings.baseurl, random.choice(Settings.fallbacks)
          for i, base_url in enumerate(base_urls):
              try:
                  url = posixpath.join(base_url, rest_of_url)
                  return getUrl(url=url, timeout=timeout, expected_hash=expected_hash)
  
              except (ArchiveDownloadError, ArchiveConnectionError) as e:
                  if i == len(base_urls) - 1:
                      raise e from e
                  else:
                      getLogger("aqt.metadata").debug(
                          f"Connection to '{base_url}' failed. Retrying with fallback '{base_urls[i + 1]}'."
                      )
  
      @staticmethod
      def iterate_folders(html_doc: str, filter_category: str = "") -> Generator[str, None, None]:
          def table_row_to_folder(tr: bs4.element.Tag) -> str:
              try:
                  return tr.find_all("td")[1].a.contents[0].rstrip("/")
              except (AttributeError, IndexError):
                  return ""
  
          soup: bs4.BeautifulSoup = bs4.BeautifulSoup(html_doc, "html.parser")
          for row in soup.body.table.find_all("tr"):
              content: str = table_row_to_folder(row)
              if not content or content == "Parent Directory":
                  continue
              if content.startswith(filter_category):
                  yield content
  
      @staticmethod
      def get_versions_extensions(html_doc: str, category: str) -> Iterator[Tuple[Optional[Version], str]]:
          def folder_to_version_extension(folder: str) -> Tuple[Optional[Version], str]:
              components = folder.split("_", maxsplit=2)
              ext = "" if len(components) < 3 else components[2]
              ver = "" if len(components) < 2 else components[1]
              return (
                  get_semantic_version(qt_ver=ver, is_preview="preview" in ext),
                  ext,
              )
  
          return map(
              folder_to_version_extension,
              MetadataFactory.iterate_folders(html_doc, category),
          )
  
      @staticmethod
      def _has_nonempty_downloads(element: Element) -> bool:
          """Returns True if the element has a nonempty '<DownloadableArchives/>' tag"""
          downloads = element.find("DownloadableArchives")
          update_file = element.find("UpdateFile")
          if downloads is None or update_file is None:
              return False
          uncompressed_size = int(update_file.attrib["UncompressedSize"])
          return downloads.text and uncompressed_size >= Settings.min_module_size
  
      def _get_qt_version_str(self, version: Version) -> str:
          """Returns a Qt version, without dots, that works in the Qt repo urls and Updates.xml files"""
          # NOTE: The url at `<base>/<host>/<target>/qt5_590/` does not exist; the real one is `qt5_59`
          patch = (
              ""
              if version.prerelease or self.archive_id.is_preview() or version in SimpleSpec("5.9.0")
              else str(version.patch)
          )
          return f"{version.major}{version.minor}{patch}"
  
      def _fetch_module_metadata(self, folder: str, predicate: Optional[Callable[[Element], bool]] = None):
          rest_of_url = posixpath.join(self.archive_id.to_url(), folder, "Updates.xml")
          xml = self.fetch_http(rest_of_url)
          return xml_to_modules(
              xml,
              predicate=predicate if predicate else MetadataFactory._has_nonempty_downloads,
          )
  
      def fetch_modules(self, version: Version, arch: str) -> List[str]:
          """Returns list of modules"""
          self.validate_extension(version)
          qt_ver_str = self._get_qt_version_str(version)
          # Example: re.compile(r"^(preview\.)?qt\.(qt5\.)?590\.(.+)$")
          pattern = re.compile(r"^(preview\.)?qt\.(qt" + str(version.major) + r"\.)?" + qt_ver_str + r"\.(.+)$")
          modules_meta = self._fetch_module_metadata(self.archive_id.to_folder(qt_ver_str))
  
          def to_module_arch(name: str) -> Tuple[Optional[str], Optional[str]]:
              _match = pattern.match(name)
              if not _match:
                  return None, None
              module_with_arch = _match.group(3)
              if self.archive_id.is_no_arch() or "." not in module_with_arch:
                  return module_with_arch, None
              module, arch = module_with_arch.rsplit(".", 1)
              if module.startswith("addons."):
                  module = module[len("addons.") :]
              return module, arch
  
          modules = set()
          for name in modules_meta.keys():
              module, _arch = to_module_arch(name)
              if _arch == arch:
                  modules.add(module)
          return sorted(modules)
  
      def fetch_modules_sde(self, cmd_type: str, version: Version) -> List[str]:
          """Returns list of modules for src/doc/examples"""
          assert (
              cmd_type in ("doc", "examples") and self.archive_id.target == "desktop"
          ), "Internal misuse of fetch_modules_sde"
          self.validate_extension(version)
          qt_ver_str = self._get_qt_version_str(version)
          modules_meta = self._fetch_module_metadata(self.archive_id.to_folder(qt_ver_str))
          # pattern: Match all names "qt.qt5.12345.doc.(\w+)
          pattern = re.compile(r"^qt\.(qt" + str(version.major) + r"\.)?" + qt_ver_str + r"\." + cmd_type + r"\.(.+)$")
  
          modules: List[str] = []
          for name in modules_meta:
              _match = pattern.match(name)
              if _match:
                  modules.append(_match.group(2))
          return modules
  
      def fetch_archives_sde(self, cmd_type: str, version: Version) -> List[str]:
          """Returns list of archives for src/doc/examples"""
          assert (
              cmd_type in ("src", "doc", "examples") and self.archive_id.target == "desktop"
          ), "Internal misuse of fetch_archives_sde"
          return self.fetch_archives(version, cmd_type, [])
  
      def fetch_archives(self, version: Version, arch: str, modules: List[str]) -> List[str]:
          qt_version_str = self._get_qt_version_str(version)
          nonempty = MetadataFactory._has_nonempty_downloads
  
          def all_modules(element: Element) -> bool:
              _module, _arch = element.find("Name").text.split(".")[-2:]
              return _arch == arch and _module != qt_version_str and nonempty(element)
  
          def specify_modules(element: Element) -> bool:
              _module, _arch = element.find("Name").text.split(".")[-2:]
              return _arch == arch and _module in modules and nonempty(element)
  
          def no_modules(element: Element) -> bool:
              name: Optional[str] = element.find("Name").text
              return name and name.endswith(f".{qt_version_str}.{arch}") and nonempty(element)
  
          predicate = no_modules if not modules else all_modules if "all" in modules else specify_modules
          try:
              mod_metadata = self._fetch_module_metadata(self.archive_id.to_folder(qt_version_str), predicate=predicate)
          except (AttributeError,) as e:
              raise ArchiveListError(f"Downloaded metadata is corrupted. {e}") from e
  
          # Did we find all requested modules?
          if modules and "all" not in modules:
              requested_set = set(modules)
              actual_set = set([_name.split(".")[-2] for _name in mod_metadata.keys()])
              not_found = sorted(requested_set.difference(actual_set))
              if not_found:
                  raise CliInputError(
                      f"The requested modules were not located: {not_found}", suggested_action=suggested_follow_up(self)
                  )
  
          csv_lists = [mod["DownloadableArchives"] for mod in mod_metadata.values()]
          return sorted(set([arc.split("-")[0] for csv in csv_lists for arc in csv.split(", ")]))
  
      def describe_filters(self) -> str:
          if self.spec is None:
              return str(self.archive_id)
          return "{} with spec {}".format(self.archive_id, self.spec)
  
  
  def suggested_follow_up(meta: MetadataFactory) -> List[str]:
      """Makes an informed guess at what the user got wrong, in the event of an error."""
      msg = []
      list_cmd = "list-tool" if meta.archive_id.is_tools() else "list-qt"
      base_cmd = "aqt {0} {1.host} {1.target}".format(list_cmd, meta.archive_id)
      versions_msg = f"Please use '{base_cmd}' to show versions of Qt available."
      arches_msg = f"Please use '{base_cmd} --arch <QT_VERSION>' to show architectures available."
      if meta.archive_id.extension:
          msg.append(f"Please use '{base_cmd} --extensions <QT_VERSION>' to list valid extensions.")
  
      if meta.archive_id.is_tools() and meta.request_type == "tool variant names":
          msg.append(f"Please use '{base_cmd}' to check what tools are available.")
      elif meta.spec is not None:
          msg.append(
              f"Please use '{base_cmd}' to check that versions of {meta.archive_id.category} "
              f"exist within the spec '{meta.spec}'."
          )
      elif meta.request_type in ("architectures", "modules", "extensions"):
          msg.append(f"Please use '{base_cmd}' to show versions of Qt available.")
          if meta.request_type == "modules":
              msg.append(f"Please use '{base_cmd} --arch <QT_VERSION>' to list valid architectures.")
      elif meta.request_type == "archives for modules":
          msg.extend([versions_msg, arches_msg, f"Please use '{base_cmd} --modules <QT_VERSION>' to show modules available."])
      elif meta.request_type == "archives for qt":
          msg.extend([versions_msg, arches_msg])
  
      return msg
  
  
  def show_list(meta: MetadataFactory):
      try:
          output = meta.getList()
          if not output:
              raise EmptyMetadata(
                  f"No {meta.request_type} available for this request.", suggested_action=suggested_follow_up(meta)
              )
          if isinstance(output, Versions):
              print(format(output))
          elif isinstance(output, ToolData):
              width: int = shutil.get_terminal_size((0, 40)).columns
              if width == 0:  # notty ?
                  print(format(output, "{:0t}"))
              elif width < 95:  # narrow terminal
                  print(format(output, "{:T}"))
              else:
                  print("{0:{1}t}".format(output, width))
          elif meta.archive_id.is_tools():
              print(*output, sep="\n")
          else:
              print(*output, sep=" ")
      except (ArchiveDownloadError, ArchiveConnectionError) as e:
          e.suggested_action = suggested_follow_up(meta)
          raise e from e
  
files.old_path: aqt/metadata.py