files: #!/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.
          """
         
  
  [... the rest is omitted ...]
files: aqt/archives.py
files: #!/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
  
  [... the rest is omitted ...]
files: aqt/archives.py
files: #!/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
  
  [... the rest is omitted ...]
files: aqt/helper.py
files: #!/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
  
  
  [... the rest is omitted ...]
files: aqt/helper.py
files: #!/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:
  
  
  [... the rest is omitted ...]
files: aqt/installer.py
files: #!/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))
  
  
  [... the rest is omitted ...]
files: aqt/installer.py
files: #!/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
  
  [... the rest is omitted ...]
files: aqt/metadata.py
files: #!/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 = (
          "was
  
  [... the rest is omitted ...]
files: aqt/metadata.py