feature: cambio merge y tag para que el changelog se genere al crear un tag nuevo

This commit is contained in:
jt
2025-11-19 20:43:22 -03:00
parent 30c6c36beb
commit be87c2098b
15 changed files with 1036 additions and 233 deletions
View File
+224
View File
@@ -0,0 +1,224 @@
from argparse import Namespace, ArgumentParser
from abc import ABC, abstractmethod
import os
from os.path import isfile
from git_flow import (
BRANCH_TYPES,
REPOSITORY_TOKEN_FILENAME,
GitFlowError,
)
from git_flow.git import FLOWCONFIG_FILE, Git
from git_flow.remote.base import RemoteAPI
from git_flow.remote.bitbucket import BitbucketRemoteAPI
from git_flow.remote.github import GithubRemoteAPI
COLOR_BLACK = "\033[30m"
COLOR_RED = "\033[31m"
COLOR_GREEN = "\033[32m"
COLOR_YELLOW = "\033[33m"
COLOR_BLUE = "\033[34m"
COLOR_5 = "\033[35m"
COLOR_6 = "\033[36m"
COLOR_7 = "\033[37m"
COLOR_BLACK_BOLD = "\033[1;30m"
COLOR_RESET = "\033[0m"
TYPE_SUCCESS = 0
TYPE_WARNING = 1
TYPE_ERROR = 2
TYPE_INFO = 3
class Command(ABC):
@abstractmethod
def name(self) -> str:
pass
@abstractmethod
def description(self) -> str:
pass
@abstractmethod
def run(self, args: Namespace = Namespace()):
pass
def __init__(self) -> None:
self.flowconfig = (
Git.get_config(FLOWCONFIG_FILE) if isfile(FLOWCONFIG_FILE) else {}
)
def success(self, msg: str):
self._print(TYPE_SUCCESS, msg)
def warning(self, msg: str):
self._print(TYPE_WARNING, msg)
def error(self, msg: str):
self._print(TYPE_ERROR, msg)
def info(self, msg: str):
self._print(TYPE_INFO, msg)
def prompt(
self,
prompt: str,
default: str | None = None,
persistent: bool = False,
strip: bool = True,
) -> str:
onetime = not persistent
while onetime or persistent:
if default:
prompt += f" [{default}]"
prompt += ": "
result = input(prompt)
result = result.strip() if strip else result
if result:
return result
elif default is not None:
return default
elif persistent:
self.error("Debe ingresar un valor no vacío.")
else:
return result
def confirm(self, question: str, default: bool = True):
suffix = " [Y/n]: " if default else " [y/N]: "
answer = input(question + suffix)
return default if len(answer) == 0 else answer.startswith("y")
def choice(self, prompt: str, options: list[str]):
print(prompt)
for i, option in enumerate(options):
print(f"\t{i+1}. {option}")
selection = None
while selection is None:
answer = input(f"Seleccione una opción [1-{len(options)}] o escribala: ")
if answer.isdigit():
answer = int(answer)
if 1 <= answer and answer <= len(options):
selection = options[answer - 1]
if not self.confirm(
f"Seleccionó la opción {answer} ({selection}), ¿es correcto?"
):
selection = None
else:
self.error(f"La opción {answer} está fuera del rango permitido.")
elif answer in options:
selection = answer
else:
self.error(f"La opción '{answer}' es inválida.")
return selection
def setup_parser(self, parser: ArgumentParser) -> ArgumentParser:
return parser
def ensure_initialized(self):
initialized = self.flowconfig["flow.initialized"] if self.flowconfig else None
if not initialized:
raise GitFlowError(
"El repositorio no fue inicializado, debe ejecutar el comando `init`."
)
elif initialized != "true":
raise GitFlowError(
f"El valor de `flow.initialized` ({initialized}) es inválido."
)
def ensure_right_branch(self):
branch = Git.get_current_branch()
if branch == "HEAD":
raise GitFlowError("No se encuentra parado sobre una rama.")
elif self.confirm(f"Se encuentra sobre la rama '{branch}'. ¿Es correcto?"):
return branch
else:
raise GitFlowError("Ejecución cancelada.")
def ensure_repository_token(self):
if not os.path.isfile(REPOSITORY_TOKEN_FILENAME):
raise GitFlowError(
"No existe un token para el repositorio en " + REPOSITORY_TOKEN_FILENAME
)
with open(REPOSITORY_TOKEN_FILENAME) as f:
return f.readline().strip()
def get_branch_env_and_type(self, branch: str) -> tuple[str, str]:
components = branch.split("/")
target_branches = self.flowconfig["flow.branches"].split(" ")
if len(components) == 2:
if components[0] not in BRANCH_TYPES:
raise GitFlowError(
f"Branch inválida, el tipo '{components[0]}' no es válido."
)
return (target_branches[0], components[0])
elif len(components) == 4:
target_branches = target_branches[1:]
if (
components[0] != "release"
or components[1] not in target_branches
or components[2] not in BRANCH_TYPES
):
raise GitFlowError(
"Branch release inválido, debe tener el siguiente formato: "
"release/<env>/<type>/<name>, pero es: " + branch
)
return (components[1], components[2])
else:
raise GitFlowError("Branch inválido: " + branch)
def get_remote_api(self, token: str) -> RemoteAPI:
if "flow.remote" not in self.flowconfig:
raise GitFlowError("El repositorio no tiene configurado un remoto.")
[host, repository] = RemoteAPI.parse(self.flowconfig["flow.remote"])
if host == "bitbucket.org":
return BitbucketRemoteAPI(repository, token)
elif host == "github.com":
return GithubRemoteAPI(repository, token)
else:
raise GitFlowError("El host del repositorio remoto es inválido")
def _print(self, type: int, msg: str):
print(self._get_color(type) + self._get_tag(type) + " " + msg + COLOR_RESET)
def _get_color(self, type: int) -> str:
if type == TYPE_SUCCESS:
return COLOR_GREEN
elif type == TYPE_WARNING:
return COLOR_YELLOW
elif type == TYPE_ERROR:
return COLOR_RED
elif type == TYPE_INFO:
return COLOR_BLUE
return ""
def _get_tag(self, type: int) -> str:
if type == TYPE_SUCCESS:
return "[success]"
elif type == TYPE_WARNING:
return "[warning]"
elif type == TYPE_ERROR:
return "[error]"
elif type == TYPE_INFO:
return "[info]"
return ""
+62
View File
@@ -0,0 +1,62 @@
from argparse import Namespace
from datetime import datetime
from git_flow import COMMIT_TYPES, GitFlowError
from git_flow.command.base import Command
from git_flow.git import Git
class CommitCommand(Command):
def name(self) -> str:
return "commit"
def description(self) -> str:
return """Crea un commit siguiendo el formato de Conventional Commits"""
def run(self, args: Namespace = Namespace()):
self.ensure_initialized()
branch = self.ensure_right_branch()
if not self._has_files_staged():
self.warning("No hay cambios en el indice para commitear.")
if self.confirm("¿Desea agregar la carpeta actual (`git add .`)?"):
Git("add", ".").exec()
else:
raise GitFlowError(
"Debe agregar algún cambio al indice para continuar."
)
commit_type = self.choice("Tipo de commit", COMMIT_TYPES)
commit_message = self.prompt(
"Mensaje (máximo recomendado: 100 caracteres)", persistent=True
)
message = commit_type + ": " + commit_message
if branch.startswith("wip/"):
original_branch = branch.removeprefix("wip/")
Git("commit", m=message).exec()
if commit_type != "wip":
suffix = datetime.now().strftime("%Y%m%dT%H%M")
Git("switch", original_branch).exec()
Git("merge", branch, squash=True).exec()
Git("commit", m=message).exec()
Git("branch", branch, "trash/" + branch + "/" + suffix, move=True).exec()
else:
if commit_type == "wip":
Git("switch", "wip/" + branch, create=True).exec()
Git("commit", m=message).exec()
def _has_files_staged(self):
status = Git.status()
if not status:
raise GitFlowError("No hay cambios para commitear.")
files_in_index = []
for s in status:
if s.worktree != "?" and s.worktree != " ":
files_in_index.append(s.file)
return len(files_in_index) > 0
+115
View File
@@ -0,0 +1,115 @@
from argparse import Namespace
import os
from git_flow import GitFlowError
from git_flow.command.base import Command
from git_flow.git import FLOWCONFIG_FILE, Git
class InitCommand(Command):
def name(self) -> str:
return "init"
def description(self) -> str:
return """Inicializa el repositorio para utilizar git-flow"""
def run(self, args: Namespace = Namespace()):
self._ensure_is_repository()
self._ensure_not_already_initialized()
branches = self._setup_flow_branches()
remote = self._setup_flow_remote()
flowconfig = {
"flow.initialized": "true",
"flow.branches": ",".join(branches),
}
if remote:
flowconfig["flow.remote"] = remote
Git.set_config(flowconfig, FLOWCONFIG_FILE)
Git("add", FLOWCONFIG_FILE).exec()
Git("commit", message="feature: initialize git-flow").exec()
self._ensure_all_flow_branches_exist(branches, remote)
def _ensure_is_repository(self):
if not Git.is_repository():
self.warning("El directorio actual no es un repositorio.")
if not self.confirm("¿Desea inicializarlo?"):
raise GitFlowError(
"No se puede continuar sin inicializar el repositorio"
)
Git("init").exec()
def _ensure_not_already_initialized(self):
if os.path.isfile(FLOWCONFIG_FILE):
flowconfig = Git.get_config(FLOWCONFIG_FILE)
if flowconfig["flow.initialized"] == "true":
raise GitFlowError(
"El repositorio ya fue inicializado para usar git-flow"
)
else:
raise GitFlowError("Valor de 'flow.initialized' es inválido")
def _setup_flow_branches(self):
self.info(
"""Ingrese las ramas que representan los entornos de deploy del proyecto en
orden creciente de cercanía al entorno productivo, y separados por coma.
Por ejemplo: "dev,test,prod"."""
)
branches = list(
map(lambda b: b.strip(), self.prompt("Ramas", "main").split(","))
)
if not all(branches):
raise GitFlowError("No puede ingresar una rama vacia")
return branches
def _setup_flow_remote(self):
self.info(
"""Puede elegir o crear un repositorio remoto para generar automáticamente PRs"""
)
remote = None
if self.confirm("¿Configurar repositorio remoto?"):
remotes = Git("remote").lines()
if not remotes:
self.info("No tiene ningún repositorio remoto, se creará uno.")
url = self.prompt(
"Ingrese la URL del repositorio (e.g. https://github.com/user/repo.git)",
persistent=True,
)
remote = "origin"
Git("remote", "add", remote, url).exec()
elif len(remotes) == 1:
remote = remotes[0]
else:
self.info("Tiene más de un remoto, seleccione el que va a utilizar.")
remote = self.choice("Remoto", remotes)
return remote
def _ensure_all_flow_branches_exist(self, branches: list[str], remote: str|None):
existing_branches = Git.get_branches()
if not all(map(lambda b: b in existing_branches, branches)):
self.warning("Algunas de las ramas de entornos expecificadas no existen.")
self.info("Debe indicar sobre que rama se crearán las ramas de entorno.")
target_branch = self.choice("Rama", existing_branches)
for branch in branches:
if branch not in existing_branches:
Git("branch", branch, target_branch).exec()
if remote:
Git("push", remote, branch, set_upstream=True).exec()
+101
View File
@@ -0,0 +1,101 @@
from argparse import Namespace
from git_flow import GitFlowError
from git_flow.changelog import Changelog
from git_flow.command.base import Command
from git_flow.git import Git
class MergeCommand(Command):
def name(self) -> str:
return "merge"
def description(self) -> str:
return """Realiza el merge de la rama, o crea el PR para hacerlo si tiene un remoto configurado"""
def run(self, args: Namespace = Namespace()):
self.ensure_initialized()
branch = self.ensure_right_branch()
(target, _) = self.get_branch_env_and_type(branch)
if "flow.remote" in self.flowconfig:
remote = self.flowconfig["flow.remote"]
token = self.ensure_repository_token()
self.run_remote(remote, token, branch, target)
else:
self.run_local(branch, target)
def run_remote(self, remote: str, token: str, branch: str, target: str):
self.check_pending_changes(True)
Git("switch", target).exec()
Git("pull").exec()
Git("switch", "-").exec()
self.check_merge_conflicts(target)
self.show_commits_to_merge(branch, target)
Git("push", remote, branch, set_upstream=True).exec()
self.create_pull_request(token, branch, target)
def run_local(self, branch: str, target: str):
self.check_pending_changes(False)
self.check_merge_conflicts(target)
self.show_commits_to_merge(branch, target)
if self.confirm(f"Mergear rama '{target}' <= '{branch}'?"):
Git("switch", target).exec()
Git("merge", branch, ff=False).exec()
def check_pending_changes(self, has_remote: bool):
status = Git.status()
if not status:
return
not_empty = any(map(lambda s: s.index != " " or s.worktree != " ", status))
if not_empty:
if not has_remote:
self.warning("Existen cambios en tu entorno de trabajo sin commitear.")
else:
raise GitFlowError("No se puede continuar con cambios pendientes.")
if not self.confirm("¿Desea continuar?", False):
raise GitFlowError("Ejecución abortada")
def show_commits_to_merge(self, branch: str, target: str):
base = Git.get_first_fork_point(branch, target)
commits = Git("log", base + "..", format="%s").lines()
if not commits:
raise GitFlowError("No hay cambios a mergear.")
print("Cambios a mergear:")
for commit in commits:
print("- " + commit)
def check_merge_conflicts(self, target: str):
merge_conflicts = Git("merge", target, ff=False, commit=False).code() != 0
if merge_conflicts:
Git("merge", abort=True).exec()
self.error(f"La rama actual tiene conflictos con {target}")
self.info(
f"""Se recomienda mergear {target} a la rama actual, resolver los conflictos localmente, y ejecutar nuevamente este comando"""
)
raise GitFlowError("Ejecución abortada.")
else:
self.success("No se detectaron merge conflicts.")
def create_pull_request(self, token: str, branch: str, target: str):
changelog = Changelog()
self.get_remote_api(token).create_pull_request(
branch,
target,
branch.replace("/", ": ").replace("-", " "),
changelog.generate_content(branch, target),
)
+49
View File
@@ -0,0 +1,49 @@
from argparse import Namespace
from git_flow import BRANCH_TYPES
from git_flow.command.base import Command
from git_flow.git import Git
class NewCommand(Command):
def name(self) -> str:
return "new"
def description(self) -> str:
return """Crea una nueva rama siguiendo conventional branches"""
def run(self, args: Namespace = Namespace()):
self.ensure_initialized()
self.ensure_right_branch()
self.info("Se creará una nueva rama de trabajo sobre la rama actual.")
self.info("Debe seleccionar el tipo de cambio a realizar.")
branch_type = self.choice("Tipos de cambio", BRANCH_TYPES)
new_branch = self._get_unique_branch_name(branch_type)
if self.confirm(f"¿Crear rama '{new_branch}' sobre la rama actual?"):
Git("switch", new_branch, create=True).exec()
def _get_unique_branch_name(self, branch_type: str) -> str:
self.info(
f"Ingrese las palabras clave que describan el cambio de tipo '{branch_type}'."
)
branch = None
branches = Git.get_branches()
while branch is None:
keywords = self.prompt(
"Palabras clave del cambio (por ejemplo: 'create worker form')",
persistent=True,
)
keywords = filter(lambda k: len(k) > 0, keywords.split(" "))
branch = branch_type + "/" + "-".join(keywords)
if branch in branches:
self.error(
f"La rama '{branch}' ya existe, cambie las palabras claves para generar otra"
)
branch = None
return branch
+114
View File
@@ -0,0 +1,114 @@
from argparse import ArgumentParser, Namespace
from git_flow import (
COMMIT_TYPE_INCREMENT,
SEMVER_MAJOR,
SEMVER_MINOR,
SEMVER_PATCH,
SEMVER_SKIP,
GitFlowError,
)
from git_flow.changelog import Changelog
from git_flow.command.base import Command
from git_flow.git import Git
class TagCommand(Command):
def name(self) -> str:
return "tag"
def description(self) -> str:
return """Crea un nuevo tag para el último merge."""
def run(self, args: Namespace = Namespace()):
self.ensure_initialized()
envs = self.flowconfig["flow.branches"].split(",")
target = Git.get_current_branch()
if target not in envs:
raise GitFlowError("Solo se pueden taggear commits en ramas principales.")
merge_commit = self.get_last_merge_commit()
self.check_not_tagged(merge_commit)
branch = Git("show", merge_commit + "^2", patch=False, format="%h").firstline()
base = Git.get_first_fork_point(branch, target)
commits = Git("log", base + ".." + branch, format="%s").lines()
next_tag = self.get_next_tag_from_commits(commits, target)
if not next_tag:
self.info(
"Los cambios realizados no implican un salto de versión, se mantiene la anterior."
)
return
(token, ci) = self.get_token_and_ci(args)
if ci:
self.get_remote_api(token).create_tag(next_tag, merge_commit)
changelog = Changelog()
changelog.update(next_tag, base, branch)
Git("add", Changelog.FILENAME).exec()
Git("commit", message=Changelog.COMMIT_MESSAGE).exec()
Git("push").exec()
else:
Git("tag", next_tag, merge_commit).exec()
def get_token_and_ci(self, args: Namespace):
try:
return (self.ensure_repository_token(), False)
except GitFlowError as e:
return (args.token, True)
def setup_parser(self, parser: ArgumentParser) -> ArgumentParser:
parser.add_argument("--token")
return parser
def get_last_merge_commit(self):
commit = Git(
"log", first_parent=True, merges=True, max_count=1, format="%h"
).firstline()
if not commit:
raise GitFlowError("No hay merges a taggear.")
return commit
def check_not_tagged(self, commit: str):
tag = (
Git("describe", commit, tags=True, exact_match=True)
.without_checking()
.firstline()
)
if tag:
raise GitFlowError("El ultimo merge ya tiene tag: " + tag)
def get_next_tag_from_commits(self, commits: list[str], branch: str) -> str | None:
increment = SEMVER_SKIP
for commit in commits:
commit_type = commit[: commit.index(":")]
increment = max(
increment, COMMIT_TYPE_INCREMENT.get(commit_type, SEMVER_SKIP)
)
tag = Git.get_current_tag() or f"v0.0.0-{branch}"
until_dash = tag.index("-") if "-" in tag else None
suffix = tag[until_dash + 1 :] if until_dash is not None else ""
[major, minor, patch] = map(int, tag[1:until_dash].split("."))
if increment == SEMVER_MAJOR:
major += 1
minor = 0
patch = 0
elif increment == SEMVER_MINOR:
minor += 1
patch = 0
elif increment == SEMVER_PATCH:
patch += 1
return None if increment == SEMVER_SKIP else f"v{major}.{minor}.{patch}{suffix}"