fix(bump_rule): add BumpRule, enum SemVerIncrement and Prerelease

commitizen/bump.py

@@ -2,60 +2,13 @@
 
 import os
 import re
-from collections import OrderedDict
 from glob import iglob
-from logging import getLogger
 from string import Template
-from typing import cast
 
-from commitizen.defaults import BUMP_MESSAGE, ENCODING, MAJOR, MINOR, PATCH
+from commitizen.defaults import BUMP_MESSAGE, ENCODING
 from commitizen.exceptions import CurrentVersionNotFoundError
-from commitizen.git import GitCommit, smart_open
-from commitizen.version_schemes import Increment, Version
-
-VERSION_TYPES = [None, PATCH, MINOR, MAJOR]
-
-logger = getLogger("commitizen")
-
-
-def find_increment(
-    commits: list[GitCommit], regex: str, increments_map: dict | OrderedDict
-) -> Increment | None:
-    if isinstance(increments_map, dict):
-        increments_map = OrderedDict(increments_map)
-
-    # Most important cases are major and minor.
-    # Everything else will be considered patch.
-    select_pattern = re.compile(regex)
-    increment: str | None = None
-
-    for commit in commits:
-        for message in commit.message.split("\n"):
-            result = select_pattern.search(message)
-
-            if result:
-                found_keyword = result.group(1)
-                new_increment = None
-                for match_pattern in increments_map.keys():
-                    if re.match(match_pattern, found_keyword):
-                        new_increment = increments_map[match_pattern]
-                        break
-
-                if new_increment is None:
-                    logger.debug(
-                        f"no increment needed for '{found_keyword}' in '{message}'"
-                    )
-
-                if VERSION_TYPES.index(increment) < VERSION_TYPES.index(new_increment):
-                    logger.debug(
-                        f"increment detected is '{new_increment}' due to '{found_keyword}' in '{message}'"
-                    )
-                    increment = new_increment
-
-                if increment == MAJOR:
-                    break
-
-    return cast(Increment, increment)
+from commitizen.git import smart_open
+from commitizen.version_schemes import Version
 
 
 def update_version_in_files(

commitizen/bump_rule.py

@@ -0,0 +1,201 @@
+from __future__ import annotations
+
+import re
+from collections.abc import Iterable, Mapping
+from enum import IntEnum, auto
+from functools import cached_property
+from typing import Callable, Protocol
+
+from commitizen.exceptions import NoPatternMapError
+
+
+class SemVerIncrement(IntEnum):
+    """An enumeration representing semantic versioning increments.
+
+    This class defines the three types of version increments according to semantic versioning:
+    - PATCH: For backwards-compatible bug fixes
+    - MINOR: For backwards-compatible functionality additions
+    - MAJOR: For incompatible API changes
+    """
+
+    PATCH = auto()
+    MINOR = auto()
+    MAJOR = auto()
+
+    def __str__(self) -> str:
+        return self.name
+
+    @classmethod
+    def safe_cast(cls, value: object) -> SemVerIncrement | None:
+        if not isinstance(value, str):
+            return None
+        try:
+            return cls[value]
+        except KeyError:
+            return None
+
+    @classmethod
+    def safe_cast_dict(cls, d: Mapping[str, object]) -> dict[str, SemVerIncrement]:
+        return {
+            k: v
+            for k, v in ((k, SemVerIncrement.safe_cast(v)) for k, v in d.items())
+            if v is not None
+        }
+
+    @staticmethod
+    def get_highest_by_messages(
+        commit_messages: Iterable[str],
+        get_increment: Callable[[str], SemVerIncrement | None],
+    ) -> SemVerIncrement | None:
+        """Find the highest version increment from a list of messages.
+
+        This function processes a list of messages and determines the highest version
+        increment needed based on the commit messages. It splits multi-line commit messages
+        and evaluates each line using the provided get_increment callable.
+
+        Args:
+            commit_messages: A list of messages to analyze.
+            get_increment: A callable that takes a commit message string and returns an
+                SemVerIncrement value (MAJOR, MINOR, PATCH) or None if no increment is needed.
+
+        Returns:
+            The highest version increment needed (MAJOR, MINOR, PATCH) or None if no
+            increment is needed. The order of precedence is MAJOR > MINOR > PATCH.
+
+        Example:
+            >>> commit_messages = ["feat: new feature", "fix: bug fix"]
+            >>> rule = ConventionalCommitBumpRule()
+            >>> SemVerIncrement.get_highest_by_messages(commit_messages, lambda x: rule.get_increment(x, False))
+            'MINOR'
+        """
+        return SemVerIncrement.get_highest(
+            get_increment(line)
+            for message in commit_messages
+            for line in message.split("\n")
+        )
+
+    @staticmethod
+    def get_highest(
+        increments: Iterable[SemVerIncrement | None],
+    ) -> SemVerIncrement | None:
+        return max(filter(None, increments), default=None)
+
+
+class BumpRule(Protocol):
+    """A protocol defining the interface for version bump rules.
+
+    This protocol specifies the contract that all version bump rule implementations must follow.
+    It defines how commit messages should be analyzed to determine the appropriate semantic
+    version increment.
+
+    The protocol is used to ensure consistent behavior across different bump rule implementations,
+    such as conventional commits or custom rules.
+    """
+
+    def get_increment(
+        self, commit_message: str, major_version_zero: bool
+    ) -> SemVerIncrement | None:
+        """Determine the version increment based on a commit message.
+
+        This method analyzes a commit message to determine what kind of version increment
+        is needed according to the Conventional Commits specification. It handles special
+        cases for breaking changes and respects the major_version_zero flag.
+
+        Args:
+            commit_message: The commit message to analyze. Should follow conventional commit format.
+            major_version_zero: If True, breaking changes will result in a MINOR version bump
+                              instead of MAJOR. This is useful for projects in 0.x.x versions.
+
+        Returns:
+            SemVerIncrement | None: The type of version increment needed:
+                - MAJOR: For breaking changes when major_version_zero is False
+                - MINOR: For breaking changes when major_version_zero is True, or for new features
+                - PATCH: For bug fixes, performance improvements, or refactors
+                - None: For commits that don't require a version bump (docs, style, etc.)
+        """
+
+
+class ConventionalCommitBumpRule(BumpRule):
+    _BREAKING_CHANGE_TYPES = set(["BREAKING CHANGE", "BREAKING-CHANGE"])
+    _MINOR_CHANGE_TYPES = set(["feat"])
+    _PATCH_CHANGE_TYPES = set(["fix", "perf", "refactor"])
+
+    def get_increment(
+        self, commit_message: str, major_version_zero: bool
+    ) -> SemVerIncrement | None:
+        if not (m := self._head_pattern.match(commit_message)):
+            return None
+
+        change_type = m.group("change_type")
+        if m.group("bang") or change_type in self._BREAKING_CHANGE_TYPES:
+            return (
+                SemVerIncrement.MINOR if major_version_zero else SemVerIncrement.MAJOR
+            )
+
+        if change_type in self._MINOR_CHANGE_TYPES:
+            return SemVerIncrement.MINOR
+
+        if change_type in self._PATCH_CHANGE_TYPES:
+            return SemVerIncrement.PATCH
+
+        return None
+
+    @cached_property
+    def _head_pattern(self) -> re.Pattern:
+        change_types = [
+            *self._BREAKING_CHANGE_TYPES,
+            *self._PATCH_CHANGE_TYPES,
+            *self._MINOR_CHANGE_TYPES,
+            "docs",
+            "style",
+            "test",
+            "build",
+            "ci",
+        ]
+        re_change_type = r"(?P<change_type>" + "|".join(change_types) + r")"
+        re_scope = r"(?P<scope>\(.+\))?"
+        re_bang = r"(?P<bang>!)?"
+        return re.compile(f"^{re_change_type}{re_scope}{re_bang}:")
+
+
+class CustomBumpRule(BumpRule):
+    def __init__(
+        self,
+        bump_pattern: str,
+        bump_map: dict[str, SemVerIncrement],
+        bump_map_major_version_zero: dict[str, SemVerIncrement],
+    ):
+        if not bump_map or not bump_pattern or not bump_map_major_version_zero:
+            raise NoPatternMapError(
+                f"Invalid bump rule: {bump_pattern=} and {bump_map=} and {bump_map_major_version_zero=}"
+            )
+
+        self.bump_pattern = re.compile(bump_pattern)
+        self.bump_map = bump_map
+        self.bump_map_major_version_zero = bump_map_major_version_zero
+
+    def get_increment(
+        self, commit_message: str, major_version_zero: bool
+    ) -> SemVerIncrement | None:
+        if not (m := self.bump_pattern.search(commit_message)):
+            return None
+
+        bump_map = (
+            self.bump_map_major_version_zero if major_version_zero else self.bump_map
+        )
+
+        try:
+            if ret := SemVerIncrement.get_highest(
+                (increment for name, increment in bump_map.items() if m.group(name)),
+            ):
+                return ret
+        except IndexError:
+            # Fallback to old school bump rule
+            pass
+
+        # Fallback to old school bump rule
+        found_keyword = m.group(1)
+        for match_pattern, increment in bump_map.items():
+            if re.match(match_pattern, found_keyword):
+                return increment
+        return None

commitizen/commands/bump.py

@@ -7,6 +7,9 @@
 import questionary
 
 from commitizen import bump, factory, git, hooks, out
+from commitizen.bump_rule import (
+    SemVerIncrement,
+)
 from commitizen.changelog_formats import get_changelog_format
 from commitizen.commands.changelog import Changelog
 from commitizen.config import BaseConfig
@@ -20,15 +23,13 @@
     InvalidManualVersion,
     NoCommitsFoundError,
     NoneIncrementExit,
-    NoPatternMapError,
     NotAGitProjectError,
     NotAllowed,
     NoVersionSpecifiedError,
 )
 from commitizen.providers import get_provider
 from commitizen.tags import TagRules
 from commitizen.version_schemes import (
-    Increment,
     InvalidVersion,
     Prerelease,
     get_version_scheme,
@@ -119,25 +120,14 @@ def is_initial_tag(
                 is_initial = questionary.confirm("Is this the first tag created?").ask()
         return is_initial
 
-    def find_increment(self, commits: list[git.GitCommit]) -> Increment | None:
+    def find_increment(self, commits: list[git.GitCommit]) -> SemVerIncrement | None:
         # Update the bump map to ensure major version doesn't increment.
-        is_major_version_zero: bool = self.bump_settings["major_version_zero"]
-        # self.cz.bump_map = defaults.bump_map_major_version_zero
-        bump_map = (
-            self.cz.bump_map_major_version_zero
-            if is_major_version_zero
-            else self.cz.bump_map
-        )
-        bump_pattern = self.cz.bump_pattern
+        is_major_version_zero = bool(self.bump_settings["major_version_zero"])
 
-        if not bump_map or not bump_pattern:
-            raise NoPatternMapError(
-                f"'{self.config.settings['name']}' rule does not support bump"
-            )
-        increment = bump.find_increment(
-            commits, regex=bump_pattern, increments_map=bump_map
+        return SemVerIncrement.get_highest_by_messages(
+            (commit.message for commit in commits),
+            lambda x: self.cz.bump_rule.get_increment(x, is_major_version_zero),
         )
-        return increment
 
     def __call__(self) -> None:  # noqa: C901
         """Steps executed to bump."""
@@ -155,8 +145,10 @@ def __call__(self) -> None:  # noqa: C901
 
         dry_run: bool = self.arguments["dry_run"]
         is_yes: bool = self.arguments["yes"]
-        increment: Increment | None = self.arguments["increment"]
-        prerelease: Prerelease | None = self.arguments["prerelease"]
+        increment: SemVerIncrement | None = SemVerIncrement.safe_cast(
+            self.arguments["increment"]
+        )
+        prerelease = Prerelease.safe_cast(self.arguments["prerelease"])
         devrelease: int | None = self.arguments["devrelease"]
         is_files_only: bool | None = self.arguments["files_only"]
         is_local_version: bool = self.arguments["local_version"]
@@ -272,7 +264,7 @@ def __call__(self) -> None:  # noqa: C901
 
             # we create an empty PATCH increment for empty tag
             if increment is None and allow_no_commit:
-                increment = "PATCH"
+                increment = SemVerIncrement.PATCH
 
             new_version = current_version.bump(
                 increment,