Skip to content
Back to top

# Source code for semantic_release.commit_parser.conventional.parser_monorepo

    from __future__ import annotations

    import os
    from fnmatch import fnmatch
    from logging import getLogger
    from pathlib import Path, PurePath, PurePosixPath, PureWindowsPath
    from re import DOTALL, compile as regexp, error as RegexError  # noqa: N812
    from typing import TYPE_CHECKING

    from semantic_release.commit_parser._base import CommitParser
    from semantic_release.commit_parser.conventional.options import (
        ConventionalCommitParserOptions,
    )
    from semantic_release.commit_parser.conventional.options_monorepo import (
        ConventionalCommitMonorepoParserOptions,
    )
    from semantic_release.commit_parser.conventional.parser import ConventionalCommitParser
    from semantic_release.commit_parser.token import (
        ParsedCommit,
        ParsedMessageResult,
        ParseError,
        ParseResult,
    )
    from semantic_release.commit_parser.util import force_str
    from semantic_release.errors import InvalidParserOptions

    if TYPE_CHECKING:  # pragma: no cover
        from git.objects.commit import Commit

    class ConventionalCommitMonorepoParser(
        CommitParser[ParseResult, ConventionalCommitMonorepoParserOptions]
    ):
        # TODO: Remove for v11 compatibility, get_default_options() will be called instead
        parser_options = ConventionalCommitMonorepoParserOptions

        def __init__(
            self, options: ConventionalCommitMonorepoParserOptions | None = None
        ) -> None:
            super().__init__(options)

            try:
                commit_scope_pattern = regexp(
                    r"\(" + self.options.scope_prefix + r"(?P<scope>[^\n]+)?\)",
                )
            except RegexError as err:
                raise InvalidParserOptions(
                    str.join(
                        "\n",
                        [
                            f"Invalid options for {self.__class__.__name__}",
                            "Unable to create regular expression from configured scope_prefix.",
                            "Please check the configured scope_prefix and remove or escape any regular expression characters.",
                        ],
                    )
                ) from err

            try:
                commit_type_pattern = regexp(
                    r"(?P<type>%s)" % str.join("|", self.options.allowed_tags)
                )
            except RegexError as err:
                raise InvalidParserOptions(
                    str.join(
                        "\n",
                        [
                            f"Invalid options for {self.__class__.__name__}",
                            "Unable to create regular expression from configured commit-types.",
                            "Please check the configured commit-types and remove or escape any regular expression characters.",
                        ],
                    )
                ) from err

            # This regular expression includes scope prefix into the pattern and forces a scope to be present
            # PSR will match the full scope but we don't include it in the scope match,
            # which implicitly strips it from being included in the returned scope.
            self._strict_scope_pattern = regexp(
                str.join(
                    "",
                    [
                        r"^" + commit_type_pattern.pattern,
                        commit_scope_pattern.pattern,
                        r"(?P<break>!)?:\s+",
                        r"(?P<subject>[^\n]+)",
                        r"(?:\n\n(?P<text>.+))?",  # commit body
                    ],
                ),
                flags=DOTALL,
            )

            self._optional_scope_pattern = regexp(
                str.join(
                    "",
                    [
                        r"^" + commit_type_pattern.pattern,
                        r"(?:\((?P<scope>[^\n]+)\))?",
                        r"(?P<break>!)?:\s+",
                        r"(?P<subject>[^\n]+)",
                        r"(?:\n\n(?P<text>.+))?",  # commit body
                    ],
                ),
                flags=DOTALL,
            )

            file_select_filters, file_ignore_filters = self._process_path_filter_options(
                self.options.path_filters
            )
            self._file_selection_filters: list[str] = file_select_filters
            self._file_ignore_filters: list[str] = file_ignore_filters

            self._logger = getLogger(
                str.join(".", [self.__module__, self.__class__.__name__])
            )

            self._base_parser = ConventionalCommitParser(
                options=ConventionalCommitParserOptions(
                    **{
                        k: getattr(self.options, k)
                        for k in ConventionalCommitParserOptions().__dataclass_fields__
                    }
                )
            )

        def get_default_options(self) -> ConventionalCommitMonorepoParserOptions:
            return ConventionalCommitMonorepoParserOptions()

        @staticmethod
        def _process_path_filter_options(  # noqa: C901
            path_filters: tuple[str, ...],
        ) -> tuple[list[str], list[str]]:
            file_ignore_filters: list[str] = []
            file_selection_filters: list[str] = []
            unique_selection_filters: set[str] = set()
            unique_ignore_filters: set[str] = set()

            for str_path in path_filters:
                str_filter = str_path[1:] if str_path.startswith("!") else str_path
                filter_list = (
                    file_ignore_filters
                    if str_path.startswith("!")
                    else file_selection_filters
                )
                unique_cache = (
                    unique_ignore_filters
                    if str_path.startswith("!")
                    else unique_selection_filters
                )

                # Since fnmatch is not too flexible, we will expand the path filters to include the name and any subdirectories
                # as this is how gitignore is interpreted. Possible scenarios:
                # | Input      | Path Normalization | Filter List               |
                # | ---------- | ------------------ | ------------------------- |
                # | /          | /                  | /**                       | done
                # | /./        | /                  | /**                       | done
                # | /**        | /**                | /**                       | done
                # | /./**      | /**                | /**                       | done
                # | /*         | /*                 | /*                        | done
                # | .          | .                  | ./**                      | done
                # | ./         | .                  | ./**                      | done
                # | ././       | .                  | ./**                      | done
                # | ./**       | ./**               | ./**                      | done
                # | ./*        | ./*                | ./*                       | done
                # | ..         | ..                 | ../**                     | done
                # | ../        | ..                 | ../**                     | done
                # | ../**      | ../**              | ../**                     | done
                # | ../*       | ../*               | ../*                      | done
                # | ../..      | ../..              | ../../**                  | done
                # | ../../     | ../../             | ../../**                  | done
                # | ../../docs | ../../docs         | ../../docs, ../../docs/** | done
                # | src        | src                | src, src/**               | done
                # | src/       | src                | src/**                    | done
                # | src/*      | src/*              | src/*                     | done
                # | src/**     | src/**             | src/**                    | done
                # | /src       | /src               | /src, /src/**             | done
                # | /src/      | /src               | /src/**                   | done
                # | /src/**    | /src/**            | /src/**                   | done
                # | /src/*     | /src/*             | /src/*                    | done
                # | ../d/f.txt | ../d/f.txt         | ../d/f.txt, ../d/f.txt/** | done
                # This expansion will occur regardless of the negation prefix

                os_path: PurePath | PurePosixPath | PureWindowsPath = PurePath(str_filter)

                if r"\\" in str_filter:
                    # Windows paths were given so we convert them to posix paths
                    os_path = PureWindowsPath(str_filter)
                    os_path = (
                        PureWindowsPath(
                            os_path.root, *os_path.parts[1:]
                        )  # drop any drive letter
                        if os_path.is_absolute()
                        else os_path
                    )
                    os_path = PurePosixPath(os_path.as_posix())

                path_normalized = str(os_path)
                if path_normalized == str(
                    Path(".").absolute().root
                ) or path_normalized == str(Path("/**")):
                    path_normalized = "/**"

                elif path_normalized == str(Path("/*")):
                    pass

                elif path_normalized == str(Path(".")) or path_normalized == str(
                    Path("./**")
                ):
                    path_normalized = "./**"

                elif path_normalized == str(Path("./*")):
                    path_normalized = "./*"

                elif path_normalized == str(Path("..")) or path_normalized == str(
                    Path("../**")
                ):
                    path_normalized = "../**"

                elif path_normalized == str(Path("../*")):
                    path_normalized = "../*"

                elif path_normalized.endswith(("..", "../**")):
                    path_normalized = f"{path_normalized.rstrip('*')}/**"

                elif str_filter.endswith(os.sep):
                    # If the path ends with a separator, it is a directory, so we add the directory and all subdirectories
                    path_normalized = f"{path_normalized}/**"

                elif not path_normalized.endswith("*"):
                    all_subdirs = f"{path_normalized}/**"
                    if all_subdirs not in unique_cache:
                        unique_cache.add(all_subdirs)
                        filter_list.append(all_subdirs)
                    # And fall through to add the path as is

                # END IF

                # Add the normalized path to the filter list if it is not already present
                if path_normalized not in unique_cache:
                    unique_cache.add(path_normalized)
                    filter_list.append(path_normalized)

            return file_selection_filters, file_ignore_filters

        def logged_parse_error(self, commit: Commit, error: str) -> ParseError:
            self._logger.debug(error)
            return ParseError(commit, error=error)

        def parse(self, commit: Commit) -> ParseResult | list[ParseResult]:
            if self.options.ignore_merge_commits and self._base_parser.is_merge_commit(
                commit
            ):
                return self._base_parser.log_parse_error(
                    commit, "Ignoring merge commit: %s" % commit.hexsha[:8]
                )

            separate_commits: list[Commit] = (
                self._base_parser.unsquash_commit(commit)
                if self.options.parse_squash_commits
                else [commit]
            )

            # Parse each commit individually if there were more than one
            parsed_commits: list[ParseResult] = list(
                map(self.parse_commit, separate_commits)
            )

            def add_linked_merge_request(
                parsed_result: ParseResult, mr_number: str
            ) -> ParseResult:
                return (
                    parsed_result
                    if not isinstance(parsed_result, ParsedCommit)
                    else ParsedCommit(
                        **{
                            **parsed_result._asdict(),
                            "linked_merge_request": mr_number,
                        }
                    )
                )

            # TODO: improve this for other VCS systems other than GitHub & BitBucket
            # Github works as the first commit in a squash merge commit has the PR number
            # appended to the first line of the commit message
            lead_commit = next(iter(parsed_commits))

            if isinstance(lead_commit, ParsedCommit) and lead_commit.linked_merge_request:
                # If the first commit has linked merge requests, assume all commits
                # are part of the same PR and add the linked merge requests to all
                # parsed commits
                parsed_commits = [
                    lead_commit,
                    *map(
                        lambda parsed_result, mr=lead_commit.linked_merge_request: (  # type: ignore[misc]
                            add_linked_merge_request(parsed_result, mr)
                        ),
                        parsed_commits[1:],
                    ),
                ]

            elif isinstance(lead_commit, ParseError) and (
                mr_match := self._base_parser.mr_selector.search(
                    force_str(lead_commit.message)
                )
            ):
                # Handle BitBucket Squash Merge Commits (see #1085), which have non angular commit
                # format but include the PR number in the commit subject that we want to extract
                linked_merge_request = mr_match.group("mr_number")

                # apply the linked MR to all commits
                parsed_commits = [
                    add_linked_merge_request(parsed_result, linked_merge_request)
                    for parsed_result in parsed_commits
                ]

            return parsed_commits

        def parse_message(
            self, message: str, strict_scope: bool = False
        ) -> ParsedMessageResult | None:
            if (
                not (parsed_match := self._strict_scope_pattern.match(message))
                and strict_scope
            ):
                return None

            if not parsed_match and not (
                parsed_match := self._optional_scope_pattern.match(message)
            ):
                return None

            return self._base_parser.create_parsed_message_result(parsed_match)

        def parse_commit(self, commit: Commit) -> ParseResult:
            """Attempt to parse the commit message with a regular expression into a ParseResult."""
            # Multiple scenarios to consider when parsing a commit message [Truth table]:
            # =======================================================================================================
            # |    ||                         INPUTS                         ||                                     |
            # |  # ||------------------------+----------------+--------------||                Result               |
            # |    || Example Commit Message | Relevant Files | Scope Prefix ||                                     |
            # |----||------------------------+----------------+--------------||-------------------------------------|
            # |  1 || type(prefix-cli): msg  |            yes |    "prefix-" ||                        ParsedCommit |
            # |  2 || type(prefix-cli): msg  |            yes |           "" ||                        ParsedCommit |
            # |  3 || type(prefix-cli): msg  |             no |    "prefix-" ||                        ParsedCommit |
            # |  4 || type(prefix-cli): msg  |             no |           "" ||                ParseError[No files] |
            # |  5 || type(scope-cli): msg   |            yes |    "prefix-" ||                        ParsedCommit |
            # |  6 || type(scope-cli): msg   |            yes |           "" ||                        ParsedCommit |
            # |  7 || type(scope-cli): msg   |             no |    "prefix-" ||  ParseError[No files & wrong scope] |
            # |  8 || type(scope-cli): msg   |             no |           "" ||                ParseError[No files] |
            # |  9 || type(cli): msg         |            yes |    "prefix-" ||                        ParsedCommit |
            # | 10 || type(cli): msg         |            yes |           "" ||                        ParsedCommit |
            # | 11 || type(cli): msg         |             no |    "prefix-" ||  ParseError[No files & wrong scope] |
            # | 12 || type(cli): msg         |             no |           "" ||                ParseError[No files] |
            # | 13 || type: msg              |            yes |    "prefix-" ||                        ParsedCommit |
            # | 14 || type: msg              |            yes |           "" ||                        ParsedCommit |
            # | 15 || type: msg              |             no |    "prefix-" ||  ParseError[No files & wrong scope] |
            # | 16 || type: msg              |             no |           "" ||                ParseError[No files] |
            # | 17 || non-conventional msg   |            yes |    "prefix-" ||          ParseError[Invalid Syntax] |
            # | 18 || non-conventional msg   |            yes |           "" ||          ParseError[Invalid Syntax] |
            # | 19 || non-conventional msg   |             no |    "prefix-" ||          ParseError[Invalid Syntax] |
            # | 20 || non-conventional msg   |             no |           "" ||          ParseError[Invalid Syntax] |
            # =======================================================================================================

            # Initial Logic Flow:
            # [1] When there are no relevant files and a scope prefix is defined, we enforce a strict scope
            # [2] When there are no relevant files and no scope prefix is defined, we parse scoped or unscoped commits
            # [3] When there are relevant files, we parse scoped or unscoped commits regardless of any defined prefix
            has_relevant_changed_files = self._has_relevant_changed_files(commit)
            strict_scope = bool(
                not has_relevant_changed_files and self.options.scope_prefix
            )
            pmsg_result = self.parse_message(
                message=force_str(commit.message),
                strict_scope=strict_scope,
            )

            if pmsg_result and (has_relevant_changed_files or strict_scope):
                self._logger.debug(
                    "commit %s introduces a %s level_bump",
                    commit.hexsha[:8],
                    pmsg_result.bump,
                )

                return ParsedCommit.from_parsed_message_result(commit, pmsg_result)

            if pmsg_result and not has_relevant_changed_files:
                return self.logged_parse_error(
                    commit,
                    f"Commit {commit.hexsha[:7]} has no changed files matching the path filter(s)",
                )

            if strict_scope and self.parse_message(str(commit.message), strict_scope=False):
                return self.logged_parse_error(
                    commit,
                    str.join(
                        " and ",
                        [
                            f"Commit {commit.hexsha[:7]} has no changed files matching the path filter(s)",
                            f"the scope does not match scope prefix '{self.options.scope_prefix}'",
                        ],
                    ),
                )

            return self.logged_parse_error(
                commit,
                f"Format Mismatch! Unable to parse commit message: {commit.message!r}",
            )

        def unsquash_commit_message(self, message: str) -> list[str]:
            return self._base_parser.unsquash_commit_message(message)

        def _has_relevant_changed_files(self, commit: Commit) -> bool:
            # Extract git root from commit
            git_root = (
                Path(commit.repo.working_tree_dir or commit.repo.working_dir)
                .absolute()
                .resolve()
            )

            cwd = Path.cwd().absolute().resolve()

            rel_cwd = cwd.relative_to(git_root) if git_root in cwd.parents else Path(".")

            sandboxed_selection_filters: list[str] = [
                str(file_filter)
                for file_filter in (
                    (
                        git_root / select_filter.rstrip("/")
                        if Path(select_filter).is_absolute()
                        else git_root / rel_cwd / select_filter
                    )
                    for select_filter in self._file_selection_filters
                )
                if git_root in file_filter.parents
            ]

            sandboxed_ignore_filters: list[str] = [
                str(file_filter)
                for file_filter in (
                    (
                        git_root / ignore_filter.rstrip("/")
                        if Path(ignore_filter).is_absolute()
                        else git_root / rel_cwd / ignore_filter
                    )
                    for ignore_filter in self._file_ignore_filters
                )
                if git_root in file_filter.parents
            ]

            # Check if the changed files of the commit that match the path filters
            for full_path in iter(
                str(git_root / rel_git_path) for rel_git_path in commit.stats.files
            ):
                # Check if the filepath matches any of the file selection filters
                if not any(
                    fnmatch(full_path, select_filter)
                    for select_filter in sandboxed_selection_filters
                ):
                    continue

                # Pass filter matches, so now evaluate if it is supposed to be ignored
                if not any(
                    fnmatch(full_path, ignore_filter)
                    for ignore_filter in sandboxed_ignore_filters
                ):
                    # No ignore filter matched, so it must be a relevant file
                    return True

            return False