PROGRAM_DESCRIPTION: str = 'Tool to help package maintainers rebase their packages to the latest upstream version'
NEW_ISSUE_LINK: str = 'https://github.com/rebase-helper/rebase-helper/issues/new'
RESULTS_DIR: str = 'rebase-helper-results'
WORKSPACE_DIR: str = 'rebase-helper-workspace'
REBASED_SOURCES_DIR: str = 'rebased-sources'
OLD_BUILD_DIR: str = 'old-build'
NEW_BUILD_DIR: str = 'new-build'
CHECKERS_DIR: str = 'checkers'
LOGS_DIR: str = 'logs'
DEBUG_LOG: str = 'debug.log'
TRACEBACK_LOG: str = 'traceback.log'
VERBOSE_LOG: str = 'verbose.log'
INFO_LOG: str = 'info.log'
REPORT: str = 'report'
CHANGES_PATCH: str = 'changes.patch'
OLD_SOURCES_DIR: str = 'old_sources'
NEW_SOURCES_DIR: str = 'new_sources'
GIT_CONFIG: str = '.gitconfig'
CONFIG_PATH: str = '$XDG_CONFIG_HOME'
CONFIG_FILENAME: str = 'rebase-helper.cfg'
ENCODING: str = locale.getpreferredencoding() class Archive:
    """Class representing an archive with sources"""

    def __init__(self, filename):
        self._filename = filename
        self._archive_type = None

        for archive_type in archive_types.values():
            if archive_type.match(self._filename):
                self._archive_type = archive_type

        if self._archive_type is None:
            raise NotImplementedError('Unsupported archive type')

    def extract_archive(self, path):
        """
        Extracts the archive into the given path

        :param path: Path where to extract the archive to.
        """
        logger.verbose('Extracting %s to %s', self._filename, path)

        try:
            archive = self._archive_type.open(self._filename)
        except (EOFError, tarfile.ReadError, lzma.LZMAError) as e:
            raise IOError(str(e)) from e

        try:
            self._archive_type.extract(archive, self._filename, path)
        finally:
            archive.close()

    @classmethod
    def get_supported_archives(cls):
        return archive_types.keys() class SpecContent:
    """Class representing content of a SPEC file."""

    # taken from build/parseSpec.c in rpm source code
    SECTION_HEADERS: List[str] = [
        '%package',
        '%prep',
        '%generate_buildrequires',
        '%build',
        '%install',
        '%check',
        '%clean',
        '%preun',
        '%postun',
        '%pretrans',
        '%posttrans',
        '%pre',
        '%post',
        '%files',
        '%changelog',
        '%description',
        '%triggerpostun',
        '%triggerprein',
        '%triggerun',
        '%triggerin',
        '%trigger',
        '%verifyscript',
        '%sepolicy',
        '%filetriggerin',
        '%filetrigger',
        '%filetriggerun',
        '%filetriggerpostun',
        '%transfiletriggerin',
        '%transfiletrigger',
        '%transfiletriggerun',
        '%transfiletriggerpostun',
        '%end',
        '%patchlist',
        '%sourcelist',
    ]

    # Comments in these sections can only be on a separate line.
    DISALLOW_INLINE_COMMENTS: List[str] = [
        '%package',
        '%patchlist',
        '%sourcelist',
        '%description',
        '%files',
        '%changelog',
    ]

    def __init__(self, content: str) -> None:
        self.sections: List[Tuple[str, List[str]]] = self._split_sections(content)

    def __str__(self) -> str:
        """Join SPEC file sections back together."""
        content = []
        for header, section in self.sections:
            if header != '%package':
                content.append(header + '\n')
            for line in section:
                content.append(line + '\n')
        return ''.join(content)

    def __getitem__(self, index: int) -> List[str]:
        return self.sections[index][1]

    @classmethod
    def get_comment_span(cls, line: str, section: str) -> Tuple[int, int]:
        """Gets span of a comment depending on the section.

        Args:
            line: Line to find the comment in.
            section: Section the line is in.

        Returns:
            Span of the comment. If no comment is found, both tuple elements
            are equal to the length of the line for convenient use in a slice.

        """
        inline_comment_allowed = not any(section.startswith(s) for s in cls.DISALLOW_INLINE_COMMENTS)
        comment = re.search(r" #.*" if inline_comment_allowed else r"^\s*#.*", line)
        return comment.span() if comment else (len(line), len(line))

    def section(self, name: str) -> Optional[List[str]]:
        """Gets content of a section.

        In case there are multiple sections with the same name, the first one is returned.

        Args:
            name: Section name.

        Returns:
            Section content as a list of lines.

        """
        for header, section in self.sections:
            if header.lower() == name.lower():
                return section
        return None

    def replace_section(self, name: str, content: List[str]) -> bool:
        """Replaces content of a section.

        In case there are multiple sections with the same name, the first one is replaced.

        Args:
            name: Section name.
            content: Section content as a list of lines.

        Returns:
            False if section was not found else True.

        """
        for i, (header, _) in enumerate(self.sections):
            if header.lower() == name.lower():
                self.sections[i] = (header, content)
                return True
        return False

    @classmethod
    def _split_sections(cls, content: str) -> List[Tuple[str, List[str]]]:
        """Splits content of a SPEC file into sections.

        Args:
            content: Content of the SPEC file

        Returns:
            The split sections represented as a list of tuples (the first element
            is the section name, the second is the content of the section).

        """
        lines = content.splitlines()
        section_headers_re = [re.compile(r'^{0}\b.*'.format(re.escape(x)), re.IGNORECASE) for x in cls.SECTION_HEADERS]

        section_beginnings: List[Optional[int]] = []
        for i, line in enumerate(lines):
            if line.startswith('%'):
                for header in section_headers_re:
                    if header.match(line):
                        section_beginnings.append(i)
        section_beginnings.append(None)

        sections = [('%package', lines[:section_beginnings[0]])]

        for i in range(len(section_beginnings) - 1):
            # Only the last element of section_beginnings is None, section_beginnings[i] can never be None
            # (the last element isn't iterated over), so section_beginnings[i] + 1 is always valid. Ignore the type.
            start = section_beginnings[i] + 1  # type: ignore
            end = section_beginnings[i + 1]
            sections.append((lines[start - 1], lines[start:end]))
        return sections class Tag:
    def __init__(self, section_index: int, section_name: str, line: int, name: str, value_span: Tuple[int, int],
                 valid: bool, index: Optional[int] = None) -> None:
        self.section_index: int = section_index
        self.section_name: str = section_name
        self.line: int = line
        self.name: str = name
        self.value_span: Tuple[int, int] = value_span
        self.valid: bool = valid
        self.index: Optional[int] = index

    def __eq__(self, other):
        return (self.section_index == other.section_index and
                self.section_name == other.section_name and
                self.line == other.line and
                self.name == other.name and
                self.value_span == other.value_span and
                self.valid == other.valid and
                self.index == other.index)

    def __str__(self) -> str:
        return (
            "Tag("
            f"section_index='{self.section_index}', "
            f"section_name='{self.section_name}', "
            f"line='{self.line}', "
            f"name='{self.name}', "
            f"value_span='{self.value_span}', "
            f"valid='{self.valid}', "
            f"index='{self.index}')"
        ) class Tags(collections.abc.Sequence):
    def __init__(self, raw_content: SpecContent, parsed_content: SpecContent) -> None:
        self.items: Tuple[Tag] = self._parse(raw_content, parsed_content)

    def __getitem__(self, index):
        return self.items[index]

    def __len__(self):
        return len(self.items)

    @classmethod
    def _map_sections_to_parsed(cls, raw_content: SpecContent, parsed_content: SpecContent) -> Dict[int, int]:
        """Creates a mapping of raw sections to parsed sections.

        Returns:
            A dict where the keys are indexes of sections in raw content and values are their
            counterparts in parsed content. Value is set to -1 if the section is not in
            parsed SpecContent.
        """
        result: Dict[int, int] = {}
        parsed = 0
        for index, (section, _) in enumerate(raw_content.sections):
            if parsed < len(parsed_content.sections) and parsed_content.sections[parsed][0].lower() == section.lower():
                result[index] = parsed
                parsed += 1
            else:
                result[index] = -1

        return result

    @classmethod
    def _parse(cls, raw_content: SpecContent, parsed_content: SpecContent) -> Tuple[Tag]:
        result: List[Tag] = []
        parsed_mapping = cls._map_sections_to_parsed(raw_content, parsed_content)
        next_source_index = 0
        next_patch_index = 0
        # counts number of occurrences of one particular section
        for section_index, (section, section_content) in enumerate(raw_content.sections):
            section = section.lower()
            parsed_section_index = parsed_mapping[section_index]
            parsed: List[str] = [] if parsed_section_index == -1 else parsed_content.sections[parsed_section_index][1]
            if section.startswith('%package'):
                tags, next_source_index, next_patch_index = cls._parse_package_tags(section, section_content, parsed,
                                                                                    section_index, next_source_index,
                                                                                    next_patch_index)
                result.extend(tags)
            elif section.startswith('%sourcelist') or section.startswith('%patchlist'):
                tags, next_source_index, next_patch_index = cls._parse_list_tags(section, section_content, parsed,
                                                                                 section_index, next_source_index,
                                                                                 next_patch_index)
                result.extend(tags)
        return cast(Tuple[Tag], tuple(result))

    @classmethod
    def _sanitize_tag(cls, name: str, next_source_index: int,
                      next_patch_index: int) -> Tuple[str, Optional[int], int, int]:
        """Sanitizes a tag name.

        Capitalizes the tag. If the tag is numbered, strips padded zeroes.
        Implements automatic numbering of Patches and Sources (adds the number
        to their name).
        """
        name, *rest = re.split(r'(\d+)', name.lower(), 1)
        sanitized_name = name.capitalize()
        next_index = next_source_index if name == 'source' else next_patch_index
        index = int(next(iter(rest), next_index))
        if name == 'source':
            return '{}{}'.format(sanitized_name, index), index, index + 1, next_patch_index
        if name == 'patch':
            return '{}{}'.format(sanitized_name, index), index, next_source_index, index + 1
        return sanitized_name, None, next_source_index, next_patch_index

    @classmethod
    def _parse_package_tags(cls, section: str, section_content: List[str], parsed: List[str], section_index: int,
                            next_source_index: int, next_patch_index: int) -> Tuple[List[Tag], int, int]:
        """Parses all tags in a %package section and determines if they are valid.

        A tag is considered valid if it is still present after evaluating all conditions.

        Note that this is not perfect - if the same tag appears in both %if and %else blocks,
        and has the same value in both, it's impossible to tell them apart, so only the latter
        is considered valid, disregarding the actual condition.

        Returns:
              A tuple containing: a tuple of all Tags object, new next source index, new next patch index.

              Indexed tag names are sanitized, for example 'Source' is replaced with 'Source0'
              and 'Patch007' with 'Patch7'.

              Tag names are capitalized, section names are lowercase.

        """
        result = []
        tag_re = re.compile(r'^(?P<prefix>(?P<name>\w+)\s*:\s*)(?P<value>.+)$')
        for line_index, line in enumerate(section_content):
            expanded = MacroHelper.expand(line, suppress_errors=True)
            if not line or not expanded:
                continue
            valid = bool(parsed and [p for p in parsed if p == expanded.rstrip()])
            m = tag_re.match(line)
            if m:
                tag_name, tag_index, next_source_index, next_patch_index = cls._sanitize_tag(m.group('name'),
                                                                                             next_source_index,
                                                                                             next_patch_index)
                result.append(Tag(section_index, section, line_index, tag_name, m.span('value'), valid, tag_index))
                continue
            m = tag_re.match(expanded)
            if m:
                start = line.find(m.group('prefix'))
                if start < 0:
                    # tag is probably defined by a macro, just ignore it
                    continue
                # conditionalized tag
                line = line[start:].rstrip('}')  # FIXME: removing trailing braces is not very robust
                m = tag_re.match(line)
                if m:
                    span = cast(Tuple[int, int], tuple(x + start for x in m.span('value')))
                    tag_name, tag_index, next_source_index, next_patch_index = cls._sanitize_tag(m.group('name'),
                                                                                                 next_source_index,
                                                                                                 next_patch_index)
                    result.append(Tag(section_index, section, line_index, tag_name, span, valid, tag_index))

        return result, next_source_index, next_patch_index

    @classmethod
    def _parse_list_tags(cls, section: str, section_content: List[str], parsed: List[str], section_index: int,
                         next_source_index: int, next_patch_index: int) -> Tuple[List[Tag], int, int]:
        """Parses all tags in a %sourcelist or %patchlist section.

        Only parses tags that are valid (that is - are in parsed), nothing more can
        consistently be detected.

        Follows how rpm works, the new Source/Patch tags are indexed starting from
        the last parsed Source/Patch tag.

        """
        tag = 'Source' if section == '%sourcelist' else 'Patch'
        result = []
        for i, line in enumerate(section_content):
            expanded = MacroHelper.expand(line, suppress_errors=True)
            is_comment = SpecContent.get_comment_span(line, section)[0] != len(line)
            if not expanded or not line or is_comment or not [p for p in parsed if p == expanded.rstrip()]:
                continue
            tag_name, tag_index, next_source_index, next_patch_index = cls._sanitize_tag(tag, next_source_index,
                                                                                         next_patch_index)
            result.append(Tag(section_index, section, i, tag_name, (0, len(line)), True, tag_index))

        return result, next_source_index, next_patch_index

    @classmethod
    def _filter(cls, tags: Tuple[Tag], section_index: Optional[int] = None, section_name: Optional[str] = None,
                name: Optional[str] = None, valid: Optional[bool] = True) -> Iterator[Tag]:
        result = iter(tags)
        if section_index is not None:
            result = filter(lambda t: t.section_index == section_index, result)
        if section_name is not None:
            result = filter(lambda t: t.section_name == section_name.lower(), result)  # type: ignore
        if name is not None:
            result = filter(lambda t: fnmatch.fnmatchcase(t.name, name.capitalize()), result)  # type: ignore
        if valid is not None:
            result = filter(lambda t: t.valid == valid, result)
        return result

    def filter(self, section_index: Optional[int] = None, section_name: Optional[str] = None,
               name: Optional[str] = None, valid: Optional[bool] = True) -> Iterator[Tag]:
        """Filters tags based on section, name or validity. Defaults to all valid tags in all sections.

        Args:
            section_index: If specified, includes tags only from section of this index.
            section_name: If specified, includes tags only from sections of this name.
            name: If specified, includes tags matching this name. Wildcards are supported.
            valid: If specified, includes tags of this validity.

        Returns:
            Iterator of matching Tag objects.

        """
        return self._filter(self.items, section_index, section_name, name, valid) class RebaseHelperError(Exception):
    """Class representing Error raised inside rebase-helper after intentionally
    catching some expected and well known exception/error.

    """

    def __init__(self, *args, **kwargs):
        """Constructor of RebaseHelperError"""
        super().__init__()
        if not args:
            self.msg = None
        elif len(args) > 1:
            self.msg = args[0] % args[1:]
        else:
            self.msg = args[0]
        self.logfiles = kwargs.get('logfiles')

    def __str__(self):
        return str(self.msg) if self.msg else '' class DownloadError(Exception):
    """Exception indicating that download of a file failed.""" class ParseError(Exception):
    pass class LookasideCacheError(Exception):
    """Exception indicating a problem in accessing lookaside cache.""" class SilentArgumentParser(argparse.ArgumentParser):
    def __init__(self):
        super().__init__(add_help=False)

    def error(self, message):
        raise ParseError(message) class CustomLogger(logging.Logger):

    TRACE: int = logging.DEBUG + 1
    VERBOSE: int = logging.DEBUG + 2
    SUCCESS: int = logging.INFO + 5
    HEADING: int = logging.INFO + 6
    IMPORTANT: int = logging.INFO + 7

    _nameToLevel: Dict[str, int] = {
        'TRACE': TRACE,
        'VERBOSE': VERBOSE,
        'SUCCESS': SUCCESS,
        'HEADING': HEADING,
        'IMPORTANT': IMPORTANT,
    }

    def __init__(self, name, level=logging.NOTSET):
        super().__init__(name, level)

        for lev, severity in self._nameToLevel.items():
            logging.addLevelName(severity, lev)

    def __getattr__(self, level):
        severity = self._nameToLevel.get(level.upper())

        def log(message, *args, **kwargs):
            if self.isEnabledFor(severity):
                self._log(severity, message, args, **kwargs)

        if severity:
            return log

        raise AttributeError class DownloadHelper:

    """Class for downloading files and performing HTTP requests."""

    @staticmethod
    def progress(download_total, downloaded, start_time, show_size=True):
        """Prints current progress and estimated remaining time of a download to the standard output.

        Args:
            download_total (int): Total download size in bytes.
            downloaded (int): Size of the already downloaded portion of a file in bytes.
            start_time (float): Time when the download started in seconds since epoch.
            show_size (bool): Whether to show the number of downloaded bytes.

        """
        bar_width = 32
        infinite_step = 256 * 1024  # move every 256 kilobytes

        delta = time.time() - start_time

        def format_time(t):
            h, rem = divmod(int(t), 3600)
            m, s = divmod(rem, 60)
            return '{:0>2d}:{:0>2d}:{:0>2d}'.format(h, m, s)

        def format_size(s):
            units = [' ', 'K', 'M', 'G', 'T']
            i = 0
            while s >= 1024.0 and i < len(units) - 1:
                s /= 1024.0
                i += 1
            return '{:>7.2F}{}'.format(s, units[i])

        if download_total < 0:
            # infinite progress bar
            pct = ' ' * 4
            pos = int(downloaded / infinite_step) % (bar_width - 5)
            bar = '[{}<=>{}]'.format(' ' * pos, ' ' * (bar_width - 5 - pos))
            ts = ' in {}'.format(format_time(delta))
        else:
            r = float(downloaded) / float(download_total) if download_total else 0.0
            pct = '{:>3d}%'.format(int(r * 100))
            pos = int(r * (bar_width - 3))
            bar = '[{}>{}]'.format('=' * pos, ' ' * (bar_width - 3 - pos))
            ts = 'eta {}'.format(format_time(delta / r - delta) if r > 0.0 else ' ' * 7 + '?')

        size = format_size(downloaded) if show_size else ''

        # no point to log progress, write directly to stdout
        sys.stdout.write('\r{}{}  {}  {} '.format(pct, bar, size, ts))
        sys.stdout.flush()

    @staticmethod
    def request(url, **kwargs):
        """Performs an HTTP request or an FTP RETR command.

        Args:
            url (str): HTTP, HTTPS or FTP URL.
            **kwargs: Keyword arguments to be passed to requests.session.get().

        Returns:
            requests.Response: Response object.

        """

        class FTPAdapter(requests.adapters.BaseAdapter):

            def send(self, request, stream=False, timeout=None, verify=True,  # pylint: disable=unused-argument
                     cert=None, proxies=None):  # pylint: disable=unused-argument
                response = requests.models.Response()
                response.request = request
                response.connection = self
                try:
                    resp = urllib.request.urlopen(request.url)
                except urllib.error.URLError as e:
                    response.status_code = 400
                    response.reason = e.reason
                else:
                    response.status_code = 200
                    response.headers = requests.structures.CaseInsensitiveDict(getattr(resp, 'headers', {}))
                    response.raw = resp
                    response.url = resp.url
                return response

            def close(self):
                pass

        session = requests.Session()
        session.mount('ftp://', FTPAdapter())

        try:
            return session.get(url, **kwargs)
        except requests.exceptions.RequestException as e:
            logger.error('%s: %s', type(e).__name__, str(e))
            return None

    @staticmethod
    def download_file(url, destination_path, blocksize=8192):
        """Downloads a file from HTTP, HTTPS or FTP URL.

        Args:
            url (str): URL to be downloaded.
            destination_path (str): Path to where the downloaded file will be stored.
            blocksize (int): Block size in bytes.

        """
        r = DownloadHelper.request(url, stream=True)
        if r is None:
            raise DownloadError("An unexpected error occurred during the download.")

        if not 200 <= r.status_code < 300:
            raise DownloadError(r.reason)

        file_size = int(r.headers.get('content-length', -1))
        if r.headers.get('content-encoding', 'identity') != 'identity':
            # use infinite progress bar in case of compressed content, there is no
            # way to determine the uncompressed size in advance
            file_size = -1

        # file exists, check the size
        if os.path.exists(destination_path):
            if file_size < 0 or file_size != os.path.getsize(destination_path):
                logger.verbose("The destination file '%s' exists, but sizes don't match! Removing it.",
                               destination_path)
                os.remove(destination_path)
            else:
                logger.verbose("The destination file '%s' exists, and the size is correct! Skipping download.",
                               destination_path)
                return
        try:
            with open(destination_path, 'wb') as local_file:
                logger.info('Downloading file from URL %s', url)
                download_start = time.time()
                downloaded = 0

                # report progress
                DownloadHelper.progress(file_size, downloaded, download_start)

                # do the actual download
                for chunk in r.iter_content(chunk_size=blocksize):
                    downloaded += len(chunk)
                    local_file.write(chunk)

                    # report progress
                    DownloadHelper.progress(file_size, downloaded, download_start)

                sys.stdout.write('\n')
                sys.stdout.flush()
        except KeyboardInterrupt as e:
            os.remove(destination_path)
            raise e class MacroHelper:

    """Class for working with RPM macros"""

    MACROS_WHITELIST: List[str] = [
        '_bindir',
        '_datadir',
        '_includedir',
        '_infodir',
        '_initdir',
        '_libdir',
        '_libexecdir',
        '_localstatedir',
        '_mandir',
        '_sbindir',
        '_sharedstatedir',
        '_sysconfdir',
        'python2_sitelib',
        'python3_sitelib',
    ]

    @staticmethod
    def expand(s, default=None, suppress_errors=False):
        try:
            if not suppress_errors:
                return rpm.expandMacro(s)
            with ConsoleHelper.Capturer(stderr=True):
                return rpm.expandMacro(s)
        except rpm.error:
            return default

    @classmethod
    def purge_macro(cls, macro: str) -> None:
        m = '%{{{}}}'.format(macro)
        while cls.expand(m, m) != m:
            with ConsoleHelper.Capturer(stderr=True):
                rpm.delMacro(macro)

    @classmethod
    def expand_macros(cls, macros):
        """Expands values of multiple macros.

        Args:
            macros (list): List of macros to be expanded, macros are
                represented as dicts.

        Returns:
            list: List of macros with expanded values.

        """
        for macro in macros:
            macro['value'] = cls.expand(macro['value'])
        return macros

    @staticmethod
    def substitute_path_with_macros(path, macros):
        """Substitutes parts of a path with macros.

        Args:
            path (str): Path to be changed.
            macros (list): Macros which can be used as a substitution.

        Returns:
            str: Path expressed using macros.

        """
        for m in macros:
            if m['value'] and m['value'] in path:
                path = path.replace(m['value'], '%{{{}}}'.format(m['name']))

        return path

    @staticmethod
    def dump():
        """Gets list of all defined macros.

        Returns:
            list: All defined macros.

        """
        macro_re = re.compile(
            r'''
            ^\s*
            (?P<level>-?\d+)
            (?P<used>=|:)
            [ ]
            (?P<name>\w+)
            (?P<options>\(.+?\))?
            [\t]
            (?P<value>.*)
            $
            ''',
            re.VERBOSE)

        with ConsoleHelper.Capturer(stderr=True) as capturer:
            rpm.expandMacro('%dump')

        macros = []

        def add_macro(properties):
            macro = dict(properties)
            macro['used'] = macro['used'] == '='
            macro['level'] = int(macro['level'])
            if parse_version(rpm.__version__) < parse_version('4.13.90'):
                # in RPM < 4.13.90 level of some macros is decreased by 1
                if macro['level'] == -1:
                    # this could be macro with level -1 or level 0, we can not be sure
                    # so just duplicate the macro for both levels
                    macros.append(macro)
                    macro = dict(macro)
                    macro['level'] = 0
                    macros.append(macro)
                elif macro['level'] in (-14, -16):
                    macro['level'] += 1
                    macros.append(macro)
                else:
                    macros.append(macro)
            else:
                macros.append(macro)

        lines = capturer.stderr.strip().split('\n')
        # last line contains only summary
        lines.pop()
        while lines:
            line = lines.pop(0)
            # squash lines ending with \
            while line.endswith('\\'):
                line = line[:-1] + lines.pop(0)
            match = macro_re.match(line)
            if match:
                add_macro(match.groupdict())
            elif macros:
                # part of the value of the last parsed macro
                macros[-1]['value'] += '\n' + line

        return macros

    @staticmethod
    def filter(macros, **kwargs):
        """Finds all macros satisfying certain conditions.

        Args:
            macros (list): Macros to be filtered.
            **kwargs: Filters to be used.

        Returns:
            list: Macros satisfying the conditions.

        """
        def _test(macro):
            return all(macro.get(k[4:]) >= v if k.startswith('min_') else
                       macro.get(k[4:]) <= v if k.startswith('max_') else
                       macro.get(k) == v for k, v in kwargs.items())

        return [m for m in macros if _test(m)] class RpmHelper:

    """Class for working with RPM database and packages."""

    ARCHES: List[str] = []

    @staticmethod
    def is_package_installed(pkg_name=None):
        """Checks whether a package is installed.

        Args:
            pkg_name (str): Name of the package.

        Returns:
            bool: Whether the package is installed.

        """
        ts = rpm.TransactionSet()
        mi = ts.dbMatch('provides', pkg_name)
        return len(mi) > 0

    @staticmethod
    def all_packages_installed(pkg_names=None):
        """Checks if all specified packages are installed.

        Args:
            pkg_names (list): List of package names to check.

        Returns:
            bool: True if all packages are installed, False otherwise.

        """
        for pkg in pkg_names:
            if not RpmHelper.is_package_installed(pkg):
                return False
        return True

    @staticmethod
    def install_build_dependencies(spec_path=None, assume_yes=False):
        """Installs build dependencies of a package using dnf.

        Args:
            spec_path (str): Absolute path to the SPEC file.
            assume_yes (bool): Whether to automatically answer 'yes' to all questions.

        Returns:
            int: Exit code of the subprocess run.

        """
        cmd = ['dnf', 'builddep', spec_path]
        if os.geteuid() != 0:
            logger.warning("Authentication required to install build dependencies using '%s'", ' '.join(cmd))
            cmd = ['pkexec'] + cmd
        if assume_yes:
            cmd.append('-y')
        return ProcessHelper.run_subprocess(cmd)

    @staticmethod
    def get_header_from_rpm(rpm_name):
        """Gets an RPM header from the given RPM package.

        Args:
            rpm_name (str): Path to the package.

        Returns:
            RpmHeader: Header object obtained from the package.

        """
        ts = rpm.TransactionSet()
        # disable signature checking
        ts.setVSFlags(rpm._RPMVSF_NOSIGNATURES)  # pylint: disable=protected-access
        with open(rpm_name, "r", encoding=ENCODING) as f:
            hdr = ts.hdrFromFdno(f)
        return RpmHeader(hdr)

    @staticmethod
    def get_arches():
        """Gets list of all known architectures"""
        arches = ['aarch64', 'noarch', 'ppc', 'riscv64', 's390', 's390x', 'src', 'x86_64']
        macros = MacroHelper.dump()
        macros = [m for m in macros if m['name'] in ('ix86', 'arm', 'mips', 'sparc', 'alpha', 'power64')]
        for m in macros:
            arches.extend(MacroHelper.expand(m['value'], '').split())
        return arches

    @classmethod
    def split_nevra(cls, s):
        """Splits string into name, epoch, version, release and arch components"""
        regexps = [
            ('NEVRA', re.compile(r'^([^:]+)-(([0-9]+):)?([^-:]+)-(.+)\.([^.]+)$')),
            ('NEVR', re.compile(r'^([^:]+)-(([0-9]+):)?([^-:]+)-(.+)()$')),
            ('NA', re.compile(r'^([^:]+)()()()()\.([^.]+)$')),
            ('N', re.compile(r'^([^:]+)()()()()()$')),
        ]
        if not cls.ARCHES:
            cls.ARCHES = cls.get_arches()
        for pattern, regexp in regexps:
            match = regexp.match(s)
            if not match:
                continue
            name = match.group(1) or None
            epoch = match.group(3) or None
            if epoch:
                epoch = int(epoch)
            version = match.group(4) or None
            release = match.group(5) or None
            arch = match.group(6) or None
            if pattern == 'NEVRA' and arch not in cls.ARCHES:
                # unknown arch, let's assume it's actually dist
                continue
            return dict(name=name, epoch=epoch, version=version, release=release, arch=arch)
        raise RebaseHelperError('Unable to split string into NEVRA.')

    @classmethod
    def parse_spec(cls, path, flags=None):
        with open(path, 'rb') as orig:
            with tempfile.NamedTemporaryFile() as tmp:
                # remove BuildArch to workaround rpm bug
                tmp.write(b''.join(l for l in orig.readlines() if not l.startswith(b'BuildArch')))
                tmp.flush()
                capturer = None
                try:
                    with ConsoleHelper.Capturer(stderr=True) as capturer:
                        result = rpm.spec(tmp.name, flags) if flags is not None else rpm.spec(tmp.name)
                except ValueError as e:
                    output = capturer.stderr.strip().split('\n') if capturer else []
                    if len(output) == 1:
                        output = output[0]
                    raise RebaseHelperError('Failed to parse SPEC file{0}'.format(
                        ': ' + str(output) if output else '')) from e
                return result

    @classmethod
    def get_rpm_spec(cls, path: str, sourcedir: str, predefined_macros: Dict[str, str]) -> rpm.spec:
        # reset all macros and settings
        rpm.reloadConfig()
        # ensure that %{_sourcedir} macro is set to proper location
        MacroHelper.purge_macro('_sourcedir')
        rpm.addMacro('_sourcedir', sourcedir)
        # add predefined macros
        for macro, value in predefined_macros.items():
            rpm.addMacro(macro, value)
        try:
            spec = cls.parse_spec(path, flags=rpm.RPMSPEC_ANYARCH)
        except RebaseHelperError:
            # try again with RPMSPEC_FORCE flag (the default)
            spec = cls.parse_spec(path)
        return spec class RpmHeader:
    def __init__(self, hdr: rpm.hdr) -> None:
        self.hdr = hdr

    def __getattr__(self, item: str) -> Any:
        def decode(s):
            if isinstance(s, bytes):
                return s.decode(ENCODING)
            return s
        result = getattr(self.hdr, item)
        if isinstance(result, list):
            return [decode(x) for x in result]
        return decode(result) class GitHelper:

    """Class which operates with git repositories"""

    # provide fallback values if system is not configured
    GIT_USER_NAME: str = 'rebase-helper'
    GIT_USER_EMAIL: str = 'rebase-helper@localhost.local'

    @classmethod
    def get_user(cls):
        try:
            return git.cmd.Git().config('user.name', get=True, stdout_as_string=True)
        except git.GitCommandError:
            logger.warning("Failed to get configured git user name, using '%s'", cls.GIT_USER_NAME)
            return cls.GIT_USER_NAME

    @classmethod
    def get_email(cls):
        try:
            return git.cmd.Git().config('user.email', get=True, stdout_as_string=True)
        except git.GitCommandError:
            logger.warning("Failed to get configured git user email, using '%s'", cls.GIT_USER_EMAIL)
            return cls.GIT_USER_EMAIL

    @classmethod
    def run_mergetool(cls, repo):
        # we can't use GitPython here, as it doesn't allow
        # for the command to attach to stdout directly
        cwd = os.getcwd()
        try:
            os.chdir(repo.working_tree_dir)
            ProcessHelper.run_subprocess(['git', 'mergetool'])
        finally:
            os.chdir(cwd) class LookasideCacheHelper:

    """Class for downloading files from Fedora/RHEL lookaside cache"""

    rpkg_config_dir: str = '/etc/rpkg'

    @classmethod
    def _read_config(cls, tool):
        config = configparser.ConfigParser()
        config.read(os.path.join(cls.rpkg_config_dir, '{}.conf'.format(tool)))
        return dict(config.items(tool, raw=True))

    @classmethod
    def _read_sources(cls, basepath):
        line_re = re.compile(r'^(?P<hashtype>[^ ]+?) \((?P<filename>[^ )]+?)\) = (?P<hash>[^ ]+?)$')
        sources = []
        path = os.path.join(basepath, 'sources')
        if os.path.isfile(path):
            with open(path, 'r', encoding=ENCODING) as f:
                for line in f.readlines():
                    line = line.strip()
                    m = line_re.match(line)
                    if m is not None:
                        d = m.groupdict()
                    else:
                        # fall back to old format of sources file
                        hsh, filename = line.split()
                        d = dict(hash=hsh, filename=filename, hashtype='md5')
                    d['hashtype'] = d['hashtype'].lower()
                    sources.append(d)
        elif os.path.exists(path):
            logger.warning("\"sources\" is not a file, skipping parsing it")
        return sources

    @classmethod
    def _write_sources(cls, basepath, sources):
        path = os.path.join(basepath, 'sources')
        with open(path, 'w', encoding=ENCODING) as f:
            for source in sources:
                f.write('{0} ({1}) = {2}\n'.format(source['hashtype'].upper(), source['filename'], source['hash']))

    @classmethod
    def _hash(cls, filename, hashtype):
        try:
            chksum = hashlib.new(hashtype)
        except ValueError as e:
            raise LookasideCacheError('Unsupported hash type \'{}\''.format(hashtype)) from e
        with open(filename, 'rb') as f:
            chunk = f.read(8192)
            while chunk:
                chksum.update(chunk)
                chunk = f.read(8192)
        return chksum.hexdigest()

    @classmethod
    def _download_source(cls, url, package, filename, hashtype, hsh, target=None):
        if target is None:
            target = os.path.basename(filename)
        if os.path.exists(target):
            if cls._hash(target, hashtype) == hsh:
                # nothing to do
                return
            else:
                os.unlink(target)
        url = '{0}/rpms/{1}/{2}/{3}/{4}/{2}'.format(url, package, filename, hashtype, hsh)
        try:
            DownloadHelper.download_file(url, target)
        except DownloadError as e:
            raise LookasideCacheError(str(e)) from e

    @classmethod
    def download(cls, tool, basepath, package, target_dir=None):
        try:
            config = cls._read_config(tool)
            url = config['lookaside']
        except (configparser.Error, KeyError) as e:
            raise LookasideCacheError('Failed to read rpkg configuration') from e
        for source in cls._read_sources(basepath):
            target = os.path.join(target_dir, source['filename'])
            cls._download_source(url, package, source['filename'], source['hashtype'], source['hash'], target)

    @classmethod
    def _upload_source(cls, url, package, source_dir, filename, hashtype, hsh):
        class ChunkedData:
            def __init__(self, check_only, chunksize=8192):
                self.check_only = check_only
                self.chunksize = chunksize
                self.start = time.time()
                fields = [
                    ('name', package),
                    ('{}sum'.format(hashtype), hsh),
                ]
                if check_only:
                    fields.append(('filename', filename))
                else:
                    fields.append(('mtime', str(int(os.stat(filename).st_mtime))))
                    with open(path, 'rb') as f:
                        rf = RequestField('file', f.read(), filename)
                        rf.make_multipart()
                        fields.append(rf)
                self.data, content_type = encode_multipart_formdata(fields)
                self.headers = {'Content-Type': content_type}

            def __iter__(self):
                totalsize = len(self.data)
                for offset in range(0, totalsize, self.chunksize):
                    transferred = min(offset + self.chunksize, totalsize)
                    if not self.check_only:
                        DownloadHelper.progress(totalsize, transferred, self.start)
                    yield self.data[offset:transferred]

        class FakeProgress(threading.Thread):
            def __init__(self, check_only, interval=0.2):
                self.check_only = check_only
                self.interval = interval
                self.stop_event = threading.Event()
                super().__init__()

            def run(self):
                if self.check_only:
                    return
                n = 0
                start = time.time()
                while not self.stop_event.is_set():
                    DownloadHelper.progress(-1, n * 256 * 1024, start, show_size=False)
                    n += 1
                    self.stop_event.wait(self.interval)

            def stop(self):
                self.stop_event.set()
                super().join()

        def post(check_only=False):
            def _post(url, data, headers):
                try:
                    # try to authenticate using opportunistic auth first
                    return requests.post(url, data=data, headers=headers,
                                         auth=requests_gssapi.HTTPSPNEGOAuth(opportunistic_auth=True))
                except requests_gssapi.exceptions.SPNEGOExchangeError:
                    return requests.post(url, data=data, headers=headers, auth=requests_gssapi.HTTPSPNEGOAuth())
            cd = ChunkedData(check_only)
            if 'devel.redhat.com' in url:
                # the only server that properly handles chunked POST requests
                r = _post(url, cd, cd.headers)
            else:
                fp = FakeProgress(check_only)
                fp.start()
                try:
                    r = _post(url, cd.data, cd.headers)
                finally:
                    fp.stop()
            if not 200 <= r.status_code < 300:
                raise LookasideCacheError('{0}: {1}'.format(r.reason, r.text.strip()))
            return r.content

        path = os.path.join(source_dir, filename)

        try:
            state = post(check_only=True)
        except (requests.exceptions.ConnectionError, LookasideCacheError) as e:
            # just log the error and bail out
            logger.error('Attempt to upload to lookaside cache failed: %s', str(e))
            return

        if state.strip() == b'Available':
            # already uploaded
            logger.info('%s is already present in lookaside cache, not uploading', path)
            return

        logger.info('Uploading %s to lookaside cache', path)
        try:
            try:
                post()
            finally:
                sys.stdout.write('\n')
                sys.stdout.flush()
        except (requests.exceptions.ConnectionError, LookasideCacheError) as e:
            # Skip error, the rebase can continue even after a failed upload
            logger.error("Upload to lookaside cache failed: %s", str(e))

    @classmethod
    def update_sources(cls, tool, basepath, package, old_sources, new_sources, upload=True, source_dir=''):
        try:
            config = cls._read_config(tool)
            url = config['lookaside_cgi']
            hashtype = config['lookasidehash']
        except (configparser.Error, KeyError) as e:
            raise LookasideCacheError('Failed to read rpkg configuration') from e
        uploaded = []
        sources = cls._read_sources(basepath)
        for idx, src in enumerate(old_sources):
            indexes = [i for i, s in enumerate(sources) if s['filename'] == src]
            if indexes:
                filename = new_sources[idx]
                if filename == src:
                    # no change
                    continue
                hsh = cls._hash(filename, hashtype)
                if upload:
                    cls._upload_source(url, package, source_dir, filename, hashtype, hsh)
                uploaded.append(filename)
                sources[indexes[0]] = dict(hash=hsh, filename=filename, hashtype=hashtype)
        cls._write_sources(basepath, sources)
        return uploaded
import argparse
import collections
import enum
import itertools
import locale
import logging
import os
import re
import shlex
import shutil
import urllib.parse
import rpm  # type: ignore
from datetime import date
from difflib import SequenceMatcher
from operator import itemgetter
from typing import List, Optional, Pattern, Tuple, Dict, Union, cast
from rebasehelper import constants
from rebasehelper.archive import Archive
from rebasehelper.spec_content import SpecContent
from rebasehelper.tags import Tag, Tags
from rebasehelper.exceptions import RebaseHelperError, DownloadError, ParseError, LookasideCacheError
from rebasehelper.argument_parser import SilentArgumentParser
from rebasehelper.logger import CustomLogger
from rebasehelper.helpers.download_helper import DownloadHelper
from rebasehelper.helpers.macro_helper import MacroHelper
from rebasehelper.helpers.rpm_helper import RpmHelper, RpmHeader
from rebasehelper.helpers.git_helper import GitHelper
from rebasehelper.helpers.lookaside_cache_helper import LookasideCacheHelper

def get_rebase_name(dir_name, name):
    """
    Function returns a name in results directory

    :param dir_name:
    :param name:
    :return: full path to results dir with name
    """
    file_name = os.path.basename(name)
    return os.path.join(dir_name, file_name)


class PatchList(list):
    def _get_index_list(self, item):
        for x in self:
            if x.index == item.index:
                return x

    def __getitem__(self, item):
        return super().__getitem__(self._get_index_list(item))


class PatchObject:

    """Class represents set of information about patches"""

    def __init__(self, path, index, strip):
        self.path = path
        self.index = index
        self.strip = strip

    def get_patch_name(self):
        return os.path.basename(self.path)


class PackageCategory(enum.Enum):
    python: Pattern[str] = re.compile(r'^python[23]?-')
    perl: Pattern[str] = re.compile(r'^perl-')
    ruby: Pattern[str] = re.compile(r'^rubygem-')
    nodejs: Pattern[str] = re.compile(r'^nodejs-')
    php: Pattern[str] = re.compile(r'^php-')
    haskell: Pattern[str] = re.compile(r'^ghc-')
    R: Pattern[str] = re.compile(r'^R-')
    rust: Pattern[str] = re.compile(r'^rust-')


def saves(func):
    """Decorator for saving the SpecFile after a method is run."""
    def wrapper(spec, *args, **kwargs):
        func(spec, *args, **kwargs)
        spec.save()

    return wrapper


class SpecFile:

    """Class representing a SPEC file. Be aware that using SpecFile
    modifies RPM macros in global context."""

    def __init__(self, path: str, sources_location: str = '', predefined_macros: Optional[Dict[str, str]] = None,
                 lookaside_cache_preset: str = 'fedpkg', keep_comments: bool = False):
        # Initialize attributes
        self.path: str = path
        self.sources_location: str = sources_location
        self.predefined_macros: Dict[str, str] = predefined_macros or {}
        self.lookaside_cache_preset: str = lookaside_cache_preset
        self.keep_comments: bool = keep_comments
        self.prep_section: str = ''
        self.sources: List[str] = []
        self.patches: Dict[str, List[PatchObject]] = {}
        self.removed_patches: List[str] = []
        self.category: Optional[PackageCategory] = None
        self.spc: rpm.spec = RpmHelper.get_rpm_spec(self.path, self.sources_location, self.predefined_macros)
        self.header: RpmHeader = RpmHeader(self.spc.sourceHeader)
        self.spec_content: SpecContent = self._read_spec_content()
        self.tags: Tags = Tags(self.spec_content, SpecContent(self.spc.parsed))

        # Load rpm information
        self._update_data()

    def __del__(self):
        # make sure there are no leftover macros
        rpm.reloadConfig()

    def download_remote_sources(self):
        """
        Method that iterates over all sources and downloads ones, which contain URL instead of just a file.

        :return: None
        """
        try:
            # try to download old sources from Fedora lookaside cache
            LookasideCacheHelper.download(self.lookaside_cache_preset, os.path.dirname(self.path), self.header.name,
                                          self.sources_location)
        except LookasideCacheError as e:
            logger.verbose("Downloading sources from lookaside cache failed. "
                           "Reason: %s.", str(e))

        # filter out only sources with URL
        remote_files = [source for source in self.sources if bool(urllib.parse.urlparse(source).scheme)]
        # download any sources that are not yet downloaded
        for remote_file in remote_files:
            local_file = os.path.join(self.sources_location, os.path.basename(remote_file))
            if not os.path.isfile(local_file):
                logger.verbose("File '%s' doesn't exist locally, downloading it.", local_file)
                try:
                    DownloadHelper.download_file(remote_file, local_file)
                except DownloadError as e:
                    raise RebaseHelperError("Failed to download file from URL {}. "
                                            "Reason: '{}'. ".format(remote_file, str(e))) from e

    def update(self) -> None:
        # explicitly discard old instance to prevent rpm from destroying
        # "sources" and "patches" lua tables after new instance is created
        self.spc = None
        self.spc = RpmHelper.get_rpm_spec(self.path, self.sources_location, self.predefined_macros)
        self.header = RpmHeader(self.spc.sourceHeader)
        self.spec_content = self._read_spec_content()

        self.tags = Tags(self.spec_content, SpecContent(self.spc.parsed))