PK!(NPPcommitizen/__init__.pyimport logging import logging.config from colorama import init from commitizen.cz.base import BaseCommitizen init() LOGGING = { "version": 1, "disable_existing_loggers": True, "formatters": {"standard": {"format": "%(message)s"}}, "handlers": { "default": { "level": "DEBUG", "formatter": "standard", "class": "logging.StreamHandler", } }, "loggers": { "commitizen": {"handlers": ["default"], "level": "INFO", "propagate": True} }, } logging.config.dictConfig(LOGGING) __all__ = ["BaseCommitizen"] PK!dqHHcommitizen/__main__.pyfrom commitizen.cli import main if __name__ == "__main__": main() PK!]commitizen/__version__.py__version__ = "1.5.0" PK!2 оcommitizen/bump.pyimport re from collections import defaultdict from itertools import zip_longest from string import Template from packaging.version import Version from typing import List, Optional, Union from commitizen.defaults import ( MAJOR, MINOR, PATCH, bump_pattern, bump_map, bump_message, ) def find_increment( messages: List[str], regex: str = bump_pattern, increments_map: dict = bump_map ) -> Optional[str]: # Most important cases are major and minor. # Everything else will be considered patch. increments_map_default = defaultdict(lambda: None, increments_map) pattern = re.compile(regex) increment = None for message in messages: result = pattern.search(message) if not result: continue found_keyword = result.group(0) new_increment = increments_map_default[found_keyword] if new_increment == "MAJOR": increment = new_increment break elif increment == "MINOR" and new_increment == "PATCH": continue increment = new_increment return increment def prerelease_generator(current_version: str, prerelease: Optional[str] = None) -> str: """ X.YaN # Alpha release X.YbN # Beta release X.YrcN # Release Candidate X.Y # Final This function might return something like 'alpha1' but it will be handled by Version. """ if not prerelease: return "" version = Version(current_version) new_prerelease_number: int = 0 if version.is_prerelease and prerelease.startswith(version.pre[0]): prev_prerelease: int = list(version.pre)[1] new_prerelease_number = prev_prerelease + 1 pre_version = f"{prerelease}{new_prerelease_number}" return pre_version def semver_generator(current_version: str, increment: str = None) -> str: version = Version(current_version) prev_release = list(version.release) increments = [MAJOR, MINOR, PATCH] increments_version = dict(zip_longest(increments, prev_release, fillvalue=0)) # This flag means that current version # must remove its prerelease tag, # so it doesn't matter the increment. # Example: 1.0.0a0 with PATCH/MINOR -> 1.0.0 if not version.is_prerelease: if increment == MAJOR: increments_version[MAJOR] += 1 increments_version[MINOR] = 0 increments_version[PATCH] = 0 elif increment == MINOR: increments_version[MINOR] += 1 increments_version[PATCH] = 0 elif increment == PATCH: increments_version[PATCH] += 1 return str( f"{increments_version['MAJOR']}." f"{increments_version['MINOR']}." f"{increments_version['PATCH']}" ) def generate_version( current_version: str, increment: str, prerelease: Optional[str] = None ) -> Version: """Based on the given increment a proper semver will be generated. For now the rules and versioning scheme is based on python's PEP 0440. More info: https://www.python.org/dev/peps/pep-0440/ Example: PATCH 1.0.0 -> 1.0.1 MINOR 1.0.0 -> 1.1.0 MAJOR 1.0.0 -> 2.0.0 """ pre_version = prerelease_generator(current_version, prerelease=prerelease) semver = semver_generator(current_version, increment=increment) # TODO: post version # TODO: dev version return Version(f"{semver}{pre_version}") def update_version_in_files(current_version: str, new_version: str, files: list): """Change old version to the new one in every file given. Note that this version is not the tag formatted one. So for example, your tag could look like `v1.0.0` while your version in the package like `1.0.0`. """ for location in files: filepath, *regex = location.split(":", maxsplit=1) if len(regex) > 0: regex = regex[0] # Read in the file filedata = [] with open(filepath, "r") as f: for line in f: if regex: is_match = re.search(regex, line) if not is_match: filedata.append(line) continue # Replace the target string filedata.append(line.replace(current_version, new_version)) # Write the file out again with open(filepath, "w") as file: file.write("".join(filedata)) def create_tag(version: Union[Version, str], tag_format: Optional[str] = None): """The tag and the software version might be different. That's why this function exists. Example: | tag | version (PEP 0440) | | --- | ------- | | v0.9.0 | 0.9.0 | | ver1.0.0 | 1.0.0 | | ver1.0.0.a0 | 1.0.0a0 | """ if isinstance(version, str): version = Version(version) if not tag_format: return version.public major, minor, patch = version.release prerelease = "" if version.is_prerelease: prerelease = f"{version.pre[0]}{version.pre[1]}" t = Template(tag_format) return t.safe_substitute( version=version, major=major, minor=minor, patch=patch, prerelease=prerelease ) def create_commit_message( current_version: Union[Version, str], new_version: Union[Version, str], message_template: str = None, ) -> str: if message_template is None: message_template = bump_message t = Template(message_template) return t.safe_substitute(current_version=current_version, new_version=new_version) PK!>commitizen/cli.pyimport io import os import sys import logging import argparse import warnings from decli import cli from pathlib import Path from configparser import RawConfigParser, NoSectionError from commitizen import defaults, commands, out, config from commitizen.__version__ import __version__ logger = logging.getLogger(__name__) data = { "prog": "cz", "description": ( "Commitizen is a cli tool to generate conventional commits.\n" "For more information about the topic go to " "https://conventionalcommits.org/" ), "formatter_class": argparse.RawDescriptionHelpFormatter, "arguments": [ {"name": "--debug", "action": "store_true", "help": "use debug mode"}, {"name": ["-n", "--name"], "help": "use the given commitizen"}, { "name": ["--version"], "action": "store_true", "help": "get the version of the installed commitizen", }, ], "subcommands": { "title": "commands", "commands": [ { "name": "ls", "help": "show available commitizens", "func": commands.ListCz, }, { "name": ["commit", "c"], "help": "create new commit", "func": commands.Commit, }, { "name": "example", "help": "show commit example", "func": commands.Example, }, { "name": "info", "help": "show information about the cz", "func": commands.Info, }, {"name": "schema", "help": "show commit schema", "func": commands.Schema}, { "name": "bump", "help": "bump semantic version based on the git log", "func": commands.Bump, "arguments": [ { "name": "--dry-run", "action": "store_true", "help": "show output to stdout, no commit, no modified files", }, { "name": "--yes", "action": "store_true", "help": "accept automatically questions done", }, { "name": "--tag-format", "help": ( "format used to tag the commmit and read it, " "use it in existing projects, " "wrap around simple quotes" ), }, { "name": "--bump-message", "help": ( "template used to create the release commmit, " "useful when working with CI" ), }, { "name": ["--prerelease", "-pr"], "help": "choose type of prerelease", "choices": ["alpha", "beta", "rc"], }, { "name": ["--increment"], "help": "manually specify the desired increment", "choices": ["MAJOR", "MINOR", "PATCH"], }, ], }, ], }, } def main(): conf = config.read_cfg() parser = cli(data) # Show help if no arg provided if len(sys.argv) == 1: parser.print_help(sys.stderr) raise SystemExit() args = parser.parse_args() if args.name: conf.update({"name": args.name}) if args.debug: warnings.warn( "Debug will be deprecated in next major version. " "Please remove it from your scripts" ) logging.getLogger("commitizen").setLevel(logging.DEBUG) if args.version: out.line(__version__) raise SystemExit() args.func(conf, vars(args))() PK!dTŕcommitizen/cmd.pyimport subprocess from typing import NamedTuple class Command(NamedTuple): out: str err: str stdout: bytes stderr: bytes def run(cmd: str) -> Command: cmd.split() process = subprocess.Popen( cmd.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE ) stdout, stderr = process.communicate() return Command(stdout.decode(), stderr.decode(), stdout, stderr) PK!d\commitizen/commands/__init__.pyfrom .bump import Bump from .commit import Commit from .example import Example from .info import Info from .list_cz import ListCz from .schema import Schema __all__ = ("Bump", "Commit", "Example", "Info", "ListCz", "Schema") PK!"Ÿcommitizen/commands/bump.pyfrom packaging.version import Version from typing import Optional import questionary from commitizen import bump, git, config, out, factory NO_COMMITS_FOUND = 3 NO_VERSION_SPECIFIED = 4 NO_PATTERN_MAP = 7 COMMIT_FAILED = 8 TAG_FAILED = 9 class Bump: """Show prompt for the user to create a guided commit.""" def __init__(self, config: dict, arguments: dict): self.config: dict = config self.arguments: dict = arguments self.parameters: dict = { **config, **{ key: arguments[key] for key in [ "dry_run", "tag_format", "prerelease", "increment", "bump_message", ] if arguments[key] is not None }, } self.cz = factory.commiter_factory(self.config) def is_initial_tag(self, current_tag_version: str, is_yes: bool = False) -> bool: """Check if reading the whole git tree up to HEAD is needed.""" is_initial = False if not git.tag_exist(current_tag_version): if is_yes: is_initial = True else: out.info(f"Tag {current_tag_version} could not be found. ") out.info( ( "Possible causes:\n" "- version in configuration is not the current version\n" "- tag_format is missing, check them using 'git tag --list'\n" ) ) is_initial = questionary.confirm("Is this the first tag created?").ask() return is_initial def __call__(self): """Steps executed to bump.""" try: current_version_instance: Version = Version(self.parameters["version"]) except TypeError: out.error("[NO_VERSION_SPECIFIED]") out.error("Check if current version is specified in config file, like:") out.error("version = 0.4.3") raise SystemExit(NO_VERSION_SPECIFIED) # Initialize values from sources (conf) current_version: str = self.config["version"] tag_format: str = self.parameters["tag_format"] bump_commit_message: str = self.parameters["bump_message"] current_tag_version: str = bump.create_tag( current_version, tag_format=tag_format ) files: list = self.parameters["files"] dry_run: bool = self.parameters["dry_run"] is_yes: bool = self.arguments["yes"] prerelease: str = self.arguments["prerelease"] increment: Optional[str] = self.arguments["increment"] is_initial = self.is_initial_tag(current_tag_version, is_yes) commits = git.get_commits(current_tag_version, from_beginning=is_initial) # No commits, there is no need to create an empty tag. # Unless we previously had a prerelease. if not commits and not current_version_instance.is_prerelease: out.error("[NO_COMMITS_FOUND]") out.error("No new commits found.") raise SystemExit(NO_COMMITS_FOUND) if increment is None: bump_pattern = self.cz.bump_pattern bump_map = self.cz.bump_map if not bump_map or not bump_pattern: out.error(f"'{self.config['name']}' rule does not support bump") raise SystemExit(NO_PATTERN_MAP) increment = bump.find_increment( commits, regex=bump_pattern, increments_map=bump_map ) # Increment is removed when current and next version # are expected to be prereleases. if prerelease and current_version_instance.is_prerelease: increment = None new_version = bump.generate_version( current_version, increment, prerelease=prerelease ) new_tag_version = bump.create_tag(new_version, tag_format=tag_format) message = bump.create_commit_message( current_version, new_version, bump_commit_message ) # Report found information out.write(message) out.write(f"tag to create: {new_tag_version}") out.write(f"increment detected: {increment}") # Do not perform operations over files or git. if dry_run: raise SystemExit() config.set_key("version", new_version.public) bump.update_version_in_files(current_version, new_version.public, files) c = git.commit(message, args="-a") if c.err: out.error(c.err) raise SystemExit(COMMIT_FAILED) c = git.tag(new_tag_version) if c.err: out.error(c.err) raise SystemExit(TAG_FAILED) out.success("Done!") PK!]commitizen/commands/commit.pyimport questionary from commitizen import factory, out, git NO_ANSWERS = 5 COMMIT_ERROR = 6 class Commit: """Show prompt for the user to create a guided commit.""" def __init__(self, config: dict, *args): self.config: dict = config self.cz = factory.commiter_factory(self.config) def __call__(self): cz = self.cz questions = cz.questions() answers = questionary.prompt(questions) if not answers: raise SystemExit(NO_ANSWERS) m = cz.message(answers) out.info(f"\n{m}\n") c = git.commit(m) if c.err: out.error(c.err) raise SystemExit(COMMIT_ERROR) if "nothing added" in c.out or "no changes added to commit" in c.out: out.error(c.out) elif c.err: out.error(c.err) else: out.write(c.out) out.success("Commit successful!") PK!If 77commitizen/commands/example.pyfrom commitizen import factory, out class Example: """Show an example so people understands the rules.""" def __init__(self, config: dict, *args): self.config: dict = config self.cz = factory.commiter_factory(self.config) def __call__(self): out.write(self.cz.example()) PK!3))commitizen/commands/info.pyfrom commitizen import factory, out class Info: """Show in depth explanation of your rules.""" def __init__(self, config: dict, *args): self.config: dict = config self.cz = factory.commiter_factory(self.config) def __call__(self): out.write(self.cz.info()) PK!-commitizen/commands/list_cz.pyfrom commitizen import out from commitizen.cz import registry class ListCz: """List currently installed rules.""" def __init__(self, config: dict, *args): self.config: dict = config def __call__(self): out.write("\n".join(registry.keys())) PK!F  commitizen/commands/schema.pyfrom commitizen import factory, out class Schema: """Show structure of the rule.""" def __init__(self, config: dict, *args): self.config: dict = config self.cz = factory.commiter_factory(self.config) def __call__(self): out.write(self.cz.schema()) PK!6 AAcommitizen/config.pyimport configparser import json import os import warnings from typing import Optional from pathlib import Path from tomlkit import parse, exceptions from commitizen import defaults class Config: def __init__(self): self._config = defaults.settings.copy() self._path: Optional[str] = None @property def config(self): return self._config @property def path(self): return self._path def update(self, data: dict): self._config.update(data) def add_path(self, path: str): self._path = path _conf = Config() def has_pyproject() -> bool: return os.path.isfile("pyproject.toml") def read_pyproject_conf(data: str) -> dict: """We expect to have a section in pyproject looking like ``` [tool.commitizen] name = "cz_conventional_commits" ``` """ doc = parse(data) try: return doc["tool"]["commitizen"] except exceptions.NonExistentKey: return {} def read_raw_parser_conf(data: str) -> dict: """We expect to have a section like this ``` [commitizen] name = cz_jira files = [ "commitizen/__version__.py", "pyproject.toml" ] # this tab at the end is important ``` """ config = configparser.ConfigParser(allow_no_value=True) config.read_string(data) try: _data: dict = dict(config["commitizen"]) if "files" in _data: files = _data["files"] _f = json.loads(files) _data.update({"files": _f}) return _data except KeyError: return {} def load_global_conf() -> dict: home = str(Path.home()) global_cfg = os.path.join(home, ".cz") if not os.path.exists(global_cfg): return {} # global conf doesnt make sense with commitizen bump # so I'm deprecating it and won't test it message = ( "Global conf will be deprecated in next major version. " "Use per project configuration. " "Remove '~/.cz' file from your conf folder." ) warnings.simplefilter("always", DeprecationWarning) warnings.warn(message, category=DeprecationWarning) with open(global_cfg, "r") as f: data = f.read() conf = read_raw_parser_conf(data) return conf def read_cfg() -> dict: allowed_cfg_files = defaults.config_files for filename in allowed_cfg_files: config_file_exists = os.path.exists(filename) if not config_file_exists: continue with open(filename, "r") as f: data: str = f.read() if "toml" in filename: conf = read_pyproject_conf(data) else: conf = read_raw_parser_conf(data) if not conf: continue _conf.update(conf) _conf.add_path(filename) return _conf.config if not _conf.path: conf = load_global_conf() if conf: _conf.update(conf) return _conf.config def set_key(key: str, value: str) -> dict: """Set or update a key in the conf. For now only strings are supported. We use to update the version number. """ if not _conf.path: return {} if "toml" in _conf.path: with open(_conf.path, "r") as f: parser = parse(f.read()) parser["tool"]["commitizen"][key] = value with open(_conf.path, "w") as f: f.write(parser.as_string()) else: parser = configparser.ConfigParser() parser.read(_conf.path) parser["commitizen"][key] = value with open(_conf.path, "w") as f: parser.write(f) return _conf.config PK!QkMcommitizen/cz/__init__.pyimport importlib import pkgutil from commitizen.cz.conventional_commits import ConventionalCommitsCz from commitizen.cz.jira import JiraSmartCz registry = {"cz_conventional_commits": ConventionalCommitsCz, "cz_jira": JiraSmartCz} plugins = { name: importlib.import_module(name).discover_this for finder, name, ispkg in pkgutil.iter_modules() if name.startswith("cz_") } registry.update(plugins) PK!||commitizen/cz/base.pyfrom typing import Optional from abc import ABCMeta, abstractmethod class BaseCommitizen(metaclass=ABCMeta): bump_pattern: Optional[str] = None bump_map: Optional[dict] = None def __init__(self, config: dict): self.config = config @abstractmethod def questions(self) -> list: """Questions regarding the commit message.""" @abstractmethod def message(self, answers: dict) -> str: """Format your git message.""" def example(self) -> str: """Example of the commit message.""" raise NotImplementedError("Not Implemented yet") def schema(self) -> str: """Schema definition of the commit message.""" raise NotImplementedError("Not Implemented yet") def info(self) -> str: """Information about the standardized commit message.""" raise NotImplementedError("Not Implemented yet") PK!<@@.commitizen/cz/conventional_commits/__init__.pyfrom .conventional_commits import ConventionalCommitsCz # noqa PK!>JJ:commitizen/cz/conventional_commits/conventional_commits.pyimport os from commitizen.cz.base import BaseCommitizen from commitizen import defaults __all__ = ["ConventionalCommitsCz"] class NoSubjectException(Exception): ... def parse_scope(text): if not text: return "" scope = text.strip().split() if len(scope) == 1: return scope[0] return "-".join(scope) def parse_subject(text): if isinstance(text, str): text = text.strip(".").strip() if not text: raise NoSubjectException return text class ConventionalCommitsCz(BaseCommitizen): bump_pattern = defaults.bump_pattern bump_map = defaults.bump_map def questions(self) -> list: questions = [ { "type": "list", "name": "prefix", "message": "Select the type of change you are committing", "choices": [ { "value": "fix", "name": "fix: A bug fix. Correlates with PATCH in SemVer", }, { "value": "feat", "name": "feat: A new feature. Correlates with MINOR in SemVer", }, {"value": "docs", "name": "docs: Documentation only changes"}, { "value": "style", "name": ( "style: Changes that do not affect the " "meaning of the code (white-space, formatting," " missing semi-colons, etc)" ), }, { "value": "refactor", "name": ( "refactor: A code change that neither fixes " "a bug nor adds a feature" ), }, { "value": "perf", "name": "perf: A code change that improves performance", }, { "value": "test", "name": ( "test: Adding missing or correcting " "existing tests" ), }, { "value": "build", "name": ( "build: Changes that affect the build system or " "external dependencies (example scopes: pip, docker, npm)" ), }, { "value": "ci", "name": ( "ci: Changes to our CI configuration files and " "scripts (example scopes: GitLabCI)" ), }, ], }, { "type": "input", "name": "scope", "message": ( "Scope. Could be anything specifying place of the " "commit change (users, db, poll):\n" ), "filter": parse_scope, }, { "type": "input", "name": "subject", "filter": parse_subject, "message": ( "Subject. Concise description of the changes. " "Imperative, lower case and no final dot:\n" ), }, { "type": "confirm", "message": "Is this a BREAKING CHANGE? Correlates with MAJOR in SemVer", "name": "is_breaking_change", "default": False, }, { "type": "input", "name": "body", "message": ( "Body. Motivation for the change and contrast this " "with previous behavior:\n" ), }, { "type": "input", "name": "footer", "message": ( "Footer. Information about Breaking Changes and " "reference issues that this commit closes:\n" ), }, ] return questions def message(self, answers: dict) -> str: prefix = answers["prefix"] scope = answers["scope"] subject = answers["subject"] body = answers["body"] footer = answers["footer"] is_breaking_change = answers["is_breaking_change"] if scope: scope = f"({scope})" if is_breaking_change: body = f"BREAKING CHANGE: {body}" if body: body = f"\n\n{body}" if footer: footer = f"\n\n{footer}" message = f"{prefix}{scope}: {subject}{body}{footer}" return message def example(self) -> str: return ( "fix: correct minor typos in code\n" "\n" "see the issue for details on the typos fixed\n" "\n" "closes issue #12" ) def schema(self) -> str: return ( "(): \n" "\n" "(BREAKING CHANGE: )\n" "\n" "