feat(bump_rule): add BumpRule, VersionIncrement, Prerelease Enum

commitizen/bump.py

@@ -2,64 +2,18 @@
 
 import os
 import re
-from collections import OrderedDict
 from glob import iglob
-from logging import getLogger
 from string import Template
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING
 
-from commitizen.defaults import BUMP_MESSAGE, MAJOR, MINOR, PATCH
+from commitizen.defaults import BUMP_MESSAGE
 from commitizen.exceptions import CurrentVersionNotFoundError
-from commitizen.git import GitCommit, smart_open
+from commitizen.git import smart_open
 
 if TYPE_CHECKING:
     from collections.abc import Generator, Iterable
 
-    from commitizen.version_schemes import Increment, VersionProtocol
-
-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.version_schemes import VersionProtocol
 
 
 def update_version_in_files(

commitizen/bump_rule.py

@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+import re
+from functools import cached_property
+from typing import TYPE_CHECKING, Protocol
+
+from commitizen.exceptions import NoPatternMapError
+from commitizen.version_increment import VersionIncrement
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+
+# Re-export for backward compatibility with code that uses
+# ``from commitizen.bump_rule import VersionIncrement``.
+__all__ = [
+    "BumpRule",
+    "ConventionalCommitBumpRule",
+    "CustomBumpRule",
+    "VersionIncrement",
+]
+
+
+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 extract_increment(
+        self, commit_message: str, major_version_zero: bool
+    ) -> VersionIncrement:
+        """Determine the version increment based on a commit message.
+
+        This method analyzes a commit message to determine what kind of version increment
+        is needed. It handles special cases for breaking changes and respects the major_version_zero flag.
+
+        See the following subclasses for more details:
+        - ConventionalCommitBumpRule: For conventional commits
+        - CustomBumpRule: For custom bump rules
+
+        Args:
+            commit_message: The commit message to analyze.
+            major_version_zero: If True, breaking changes will result in a MINOR version bump instead of MAJOR
+
+        Returns:
+            VersionIncrement: The type of version increment needed:
+        """
+
+
+class ConventionalCommitBumpRule(BumpRule):
+    _BREAKING_CHANGE_TYPES = {"BREAKING CHANGE", "BREAKING-CHANGE"}
+    _MINOR_CHANGE_TYPES = {"feat"}
+    _PATCH_CHANGE_TYPES = {"fix", "perf", "refactor"}
+
+    def extract_increment(
+        self, commit_message: str, major_version_zero: bool
+    ) -> VersionIncrement:
+        if not (m := self._head_pattern.match(commit_message)):
+            return VersionIncrement.NONE
+
+        change_type = m.group("change_type")
+        if m.group("bang") or change_type in self._BREAKING_CHANGE_TYPES:
+            return (
+                VersionIncrement.MINOR if major_version_zero else VersionIncrement.MAJOR
+            )
+
+        if change_type in self._MINOR_CHANGE_TYPES:
+            return VersionIncrement.MINOR
+
+        if change_type in self._PATCH_CHANGE_TYPES:
+            return VersionIncrement.PATCH
+
+        return VersionIncrement.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: Mapping[str, VersionIncrement],
+        bump_map_major_version_zero: Mapping[str, VersionIncrement],
+    ) -> None:
+        """Initialize a custom bump rule for version incrementing.
+
+        This constructor creates a rule that determines how version numbers should be
+        incremented based on commit messages. It validates and compiles the provided
+        pattern and maps for use in version bumping.
+
+        The fallback logic is used for backward compatibility.
+
+        Args:
+            bump_pattern: A regex pattern string used to match commit messages.
+                Example: r"^((?P<major>major)|(?P<minor>minor)|(?P<patch>patch))(?P<scope>\\(.+\\))?(?P<bang>!)?:"
+
+                Or with fallback regex: r"^((BREAKING[\\-\\ ]CHANGE|\\w+)(\\(.+\\))?!?):"  # First group is type
+            bump_map: A mapping of commit types to their corresponding version increments.
+                Example: {
+                    "major": VersionIncrement.MAJOR,
+                    "bang": VersionIncrement.MAJOR,
+                    "minor": VersionIncrement.MINOR,
+                    "patch": VersionIncrement.PATCH
+                }
+                Or with fallback: {
+                    (r"^.+!$", VersionIncrement.MAJOR),
+                    (r"^BREAKING[\\-\\ ]CHANGE", VersionIncrement.MAJOR),
+                    (r"^feat", VersionIncrement.MINOR),
+                    (r"^fix", VersionIncrement.PATCH),
+                    (r"^refactor", VersionIncrement.PATCH),
+                    (r"^perf", VersionIncrement.PATCH),
+                }
+            bump_map_major_version_zero: A mapping of commit types to version increments
+                specifically for when the major version is 0. This allows for different
+                versioning behavior during initial development.
+                The format is the same as bump_map.
+
+        Raises:
+            NoPatternMapError: If any of the required parameters are empty or None
+        """
+        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 extract_increment(
+        self, commit_message: str, major_version_zero: bool
+    ) -> VersionIncrement:
+        if not (m := self.bump_pattern.search(commit_message)):
+            return VersionIncrement.NONE
+
+        effective_bump_map = (
+            self.bump_map_major_version_zero if major_version_zero else self.bump_map
+        )
+
+        try:
+            increments = (
+                increment
+                for name, increment in effective_bump_map.items()
+                if m.group(name)
+            )
+            increment = max(increments, default=VersionIncrement.NONE)
+            if increment != VersionIncrement.NONE:
+                return increment
+        except IndexError:
+            pass
+
+        # Fallback to legacy bump rule, for backward compatibility
+        found_keyword = m.group(1)
+        for match_pattern, increment in effective_bump_map.items():
+            if re.match(match_pattern, found_keyword):
+                return increment
+        return VersionIncrement.NONE

commitizen/commands/bump.py

@@ -7,6 +7,7 @@
 import questionary
 
 from commitizen import bump, factory, git, hooks, out
+from commitizen.bump_rule import VersionIncrement
 from commitizen.changelog_formats import get_changelog_format
 from commitizen.commands.changelog import Changelog
 from commitizen.defaults import Settings
@@ -18,14 +19,12 @@
     InvalidManualVersion,
     NoCommitsFoundError,
     NoneIncrementExit,
-    NoPatternMapError,
     NotAGitProjectError,
     NotAllowed,
 )
 from commitizen.providers import get_provider
 from commitizen.tags import TagRules
 from commitizen.version_schemes import (
-    Increment,
     InvalidVersion,
     Prerelease,
     VersionProtocol,
@@ -53,7 +52,7 @@ class BumpArgs(Settings, total=False):
     get_next: bool  # TODO: maybe rename to `next_version_to_stdout`
     git_output_to_stderr: bool
     increment_mode: str
-    increment: Increment | None
+    increment: VersionIncrement | None
     local_version: bool
     manual_version: str | None
     no_verify: bool
@@ -144,28 +143,23 @@ def _is_initial_tag(
         )
         return bool(questionary.confirm("Is this the first tag created?").ask())
 
-    def _find_increment(self, commits: list[git.GitCommit]) -> Increment | None:
+    def _find_increment(self, commits: list[git.GitCommit]) -> VersionIncrement:
         # Update the bump map to ensure major version doesn't increment.
-        # self.cz.bump_map = defaults.bump_map_major_version_zero
-        bump_map = (
-            self.cz.bump_map_major_version_zero
-            if self.bump_settings["major_version_zero"]
-            else self.cz.bump_map
-        )
-        bump_pattern = self.cz.bump_pattern
+        is_major_version_zero = 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"
-            )
-        return bump.find_increment(commits, regex=bump_pattern, increments_map=bump_map)
+        return VersionIncrement.get_highest_by_messages(
+            (commit.message for commit in commits),
+            lambda x: self.cz.bump_rule.extract_increment(x, is_major_version_zero),
+        )
 
     def _validate_arguments(self, current_version: VersionProtocol) -> None:
         errors: list[str] = []
+        increment = VersionIncrement.safe_cast(self.arguments["increment"])
+        prerelease = Prerelease.safe_cast(self.arguments["prerelease"])
         if self.arguments["manual_version"]:
             for val, option in (
-                (self.arguments["increment"], "--increment"),
-                (self.arguments["prerelease"], "--prerelease"),
+                (increment != VersionIncrement.NONE, "--increment"),
+                (prerelease, "--prerelease"),
                 (self.arguments["devrelease"] is not None, "--devrelease"),
                 (self.arguments["local_version"], "--local-version"),
                 (self.arguments["build_metadata"], "--build-metadata"),
@@ -186,8 +180,9 @@ def _validate_arguments(self, current_version: VersionProtocol) -> None:
 
     def _resolve_increment_and_new_version(
         self, current_version: VersionProtocol, current_tag: git.GitTag | None
-    ) -> tuple[Increment | None, VersionProtocol]:
-        increment = self.arguments["increment"]
+    ) -> tuple[VersionIncrement, VersionProtocol]:
+        increment = VersionIncrement.safe_cast(self.arguments["increment"])
+        prerelease = Prerelease.safe_cast(self.arguments["prerelease"])
         if manual_version := self.arguments["manual_version"]:
             try:
                 return increment, self.scheme(manual_version)
@@ -197,7 +192,7 @@ def _resolve_increment_and_new_version(
                     f"Invalid manual version: '{manual_version}'"
                 ) from exc
 
-        if increment is None:
+        if increment == VersionIncrement.NONE:
             commits = git.get_commits(current_tag.name if current_tag else None)
 
             # No commits, there is no need to create an empty tag.
@@ -214,8 +209,8 @@ def _resolve_increment_and_new_version(
         # It may happen that there are commits, but they are not eligible
         # for an increment, this generates a problem when using prerelease (#281)
         if (
-            self.arguments["prerelease"]
-            and increment is None
+            prerelease
+            and increment == VersionIncrement.NONE
             and not current_version.is_prerelease
         ):
             raise NoCommitsFoundError(
@@ -225,12 +220,12 @@ def _resolve_increment_and_new_version(
             )
 
         # we create an empty PATCH increment for empty tag
-        if increment is None and self.arguments["allow_no_commit"]:
-            increment = "PATCH"
+        if self.arguments["allow_no_commit"]:
+            increment = max(increment, VersionIncrement.PATCH)
 
         return increment, current_version.bump(
             increment,
-            prerelease=self.arguments["prerelease"],
+            prerelease=prerelease,
             prerelease_offset=self.bump_settings["prerelease_offset"],
             devrelease=self.arguments["devrelease"],
             is_local_version=self.arguments["local_version"],
@@ -277,7 +272,10 @@ def __call__(self) -> None:
 
         new_tag_version = rules.normalize_tag(new_version)
         if next_version_to_stdout:
-            if increment is None and new_tag_version == current_tag_version:
+            if (
+                increment == VersionIncrement.NONE
+                and new_tag_version == current_tag_version
+            ):
                 raise NoneIncrementExit(
                     "[NO_COMMITS_TO_BUMP]\n"
                     "The commits found are not eligible to be bumped"
@@ -290,7 +288,7 @@ def __call__(self) -> None:
         )
         # Report found information
         information = f"{message}\ntag to create: {new_tag_version}\n"
-        if increment:
+        if increment != VersionIncrement.NONE:
             information += f"increment detected: {increment}\n"
 
         if self.changelog_to_stdout:
@@ -301,7 +299,10 @@ def __call__(self) -> None:
         else:
             out.write(information)
 
-        if increment is None and new_tag_version == current_tag_version:
+        if (
+            increment == VersionIncrement.NONE
+            and new_tag_version == current_tag_version
+        ):
             raise NoneIncrementExit(
                 "[NO_COMMITS_TO_BUMP]\nThe commits found are not eligible to be bumped"
             )