refactor: cambio estructura de proyecto siguiendo guidelines de python

This commit is contained in:
jt
2025-11-03 21:46:49 -03:00
parent 44b2adecb8
commit 9e1c710373
7 changed files with 39 additions and 34 deletions
+663
View File
@@ -0,0 +1,663 @@
#!/usr/bin/env python3
import os
import sys
import datetime
import requests
import locale
from .git import Git
from .io import *
REPOSITORY_TOKEN_PATH=".repository-token"
LOCALE = 'es_AR.UTF-8'
def main():
try:
locale.setlocale(locale.LC_ALL, LOCALE)
except locale.Error as e:
print_warning(f"No se pudo configurar el locale '{LOCALE}': {e}")
args = sys.argv
if len(args) < 2:
print_error("missing command")
return
command = args[1]
try:
if command == "new":
new_command()
elif command == "init":
init_command()
elif command == "commit":
commit_command()
elif command == "merge":
merge_command()
elif command == "tag":
tag_command()
elif command == "release":
release_command()
elif command == "help":
help_command()
else:
help_command(command)
except Exception as e:
print_error(f"Ocurrio un error al ejecutar el comando: {e}")
BRANCH_TYPES = [
"feature",
"refactor",
"bugfix",
"hotfix",
"perf",
"docs",
"typo",
"testing",
"breaking",
]
BRANCH_TYPES_INCREMENT = {
"feature": "minor",
"refactor": "minor",
"bugfix": "patch",
"hotfix": "patch",
"perf": "minor",
"docs": "patch",
"typo": "patch",
"testing": "patch",
"breaking": "major",
}
COMMIT_TYPES = [
"feature",
"bugfix",
"hotfix",
"refactor",
"perf",
"docs",
"typo",
"testing",
"ignore",
"wip"
]
DEFAULT_REMOTE = "origin"
def new_command():
current_branch = ensure_right_branch()
print_info("Se creará una nueva rama de trabajo sobre la rama actual.")
print_info("Seleccione el tipo de cambio a realizar.")
branch_type = choose("Tipos de cambio", BRANCH_TYPES)
branch = get_unique_branch_name(branch_type)
if confirm(f"Crear rama '{branch}' sobre rama '{current_branch}'"):
Git("switch", branch, create=True).exec()
else:
raise RuntimeError("Abortando operación")
def init_command():
if not Git("rev-parse", is_inside_work_tree=True).code() == 0:
print_warning("El directorio actual no es un repositorio.")
if not confirm("¿Desea inicializarlo?"):
raise RuntimeError("Abortando operación")
Git("init").exec()
initialized = Git.flow_config("flow.initialized").without_checking().firstline()
if initialized == "true":
print_error("El repositorio ya fue inicializado")
return
elif initialized != "":
print_error("El valor de 'flow.initialized' ({initialized}) es inválido")
return
print_info("Ingrese las ramas principales del repositorio separadas por espacio.")
branches = input("Ramas [dev main]: ").strip()
branches = branches.split(" ") if len(branches) > 0 else ["dev", "main"]
base = None
if confirm("¿Desea configurar un repositorio remoto?"):
remote = DEFAULT_REMOTE
remotes = Git("remote").lines()
if len(remotes) == 0:
print("El repositorio no tiene ningún remoto.")
url = None
while url is None:
url = input("Ingrese la URL del repositorio (https://github.com/user/repo.git): ").strip()
if len(url) == 0:
print_error("Debe ingresar una URL")
url = None
elif len(remotes) == 1:
remote = remotes[0]
else:
print("Tiene más de un remoto, seleccione el que desea utilizar.")
remote = choose("Remotos: ", remotes)
Git.flow_config("flow.remote", remote).exec()
Git.flow_config("flow.branches", " ".join(branches)).exec()
Git.flow_config("flow.initialized", "true").exec()
Git("add", Git.FLOWCONFIG_FILE).exec()
Git("commit", m="feature: initialize git-flow").exec()
existing_branches = Git.get_branches()
while base is None:
default = Git("rev-parse", "HEAD", abbrev_ref=True).firstline()
base = input(f"Ingrese la rama sobre la que desea crear las ramas principales [{default}]: ")
if not base:
base = default
elif base not in existing_branches:
print_error("Debe ingresar una rama existente.")
base = None
if confirm(f"¿Crear las ramas ingresadas sobre '{base}'?"):
for branch in branches:
if branch in existing_branches:
continue
if base == "HEAD":
Git("branch", branch).exec()
else:
Git("branch", branch, base).exec()
def commit_command():
ensure_initialized()
branch = ensure_right_branch()
status = Git.status()
if not status:
print_error("No hay cambios para commitear.")
return
files_in_index = []
for s in status:
if s.worktree != "?" and s.worktree != " ":
files_in_index.append(s.file)
if not files_in_index:
print_warning("No hay cambios en el indice para commitear.")
if confirm("¿Desea agregar la carpeta actual (`git add .`)?"):
Git("add", ".").exec()
else:
print_error("Debe agregar algún cambio al indice para continuar.")
return
commit_type = choose("Tipo de commit", COMMIT_TYPES)
message = commit_type + ": " + get_commit_message()
if branch.startswith("wip/"):
original_branch = branch.removeprefix("wip/")
Git("commit", m=message).exec()
if commit_type != "wip":
now = datetime.datetime.now()
suffix = now.strftime("%Y-%m-%d_%H.%M.%S")
Git("switch", original_branch).exec()
Git("merge", branch, squash=True).exec()
Git("commit", m=message).exec()
Git("branch", branch, branch + "/" + suffix, move=True).exec()
else:
if commit_type == "wip":
Git("switch", "wip/" + branch, create=True).exec()
Git("commit", m=message).exec()
def merge_command():
ensure_initialized()
branch = ensure_right_branch()
if branch.startswith("wip/"):
print_error("No se pueden mergear ramas de tipo 'wip'")
return
(target, _) = get_branch_env_and_type(branch)
status = Git.status()
remote = Git.flow_config("flow.remote").firstline()
if status:
not_empty = any(map(lambda s: s.index != " " or s.worktree != " ", status))
if not_empty:
print_warning("Existen cambios en tu entorno de trabajo sin commitear.")
if not confirm("¿Desea continuar?", False):
return
if remote:
Git("switch", target).exec()
Git("pull").exec()
Git("switch", "-").exec()
commits = Git("log", target + "..", format="%s").lines()
if not commits:
print_error("No hay cambios a mergear.")
return
print("Cambios a mergear:")
for commit in commits:
if commit != Git.UPDATED_CHANGELOG_MSG:
print("- " + commit)
merge_conflicts = Git("merge", target, ff=False, commit=False).exec()
if merge_conflicts:
Git("merge", abort=True).exec()
print_error(f"La rama actual tiene conflictos con '{target}'. "
"Se recomienda mergear la rama a la actual, resolver los conflictos, "
"y ejecutar nuevamente este comando.")
return
if update_changelog(branch, target):
Git("add", Git.CHANGELOG_FILE).exec()
Git("commit", message=Git.UPDATED_CHANGELOG_MSG).exec()
if remote:
Git("push", remote, branch, set_upstream=True).exec()
url = Git("remote", "get-url", remote).firstline()
if url is not None:
[service, repository] = parse_url(url)
create_pull_request(service, repository, branch, target)
else:
if confirm(f"Mergear rama '{target}' <= '{branch}'?"):
Git("switch", target).exec()
Git("merge", branch, ff=False).exec()
def parse_url(url: str) -> list[str]:
if "@" in url:
# git@github.com:username/repository.git
start = url.find("@") + 1
end = url.find(":")
service = url[start : end]
start = end + 1
end = url.find(".git") if ".git" in url else len(url)
repository = url[start : end]
elif "://" in url:
# https://github.com/username/repository.git
start = url.index("://") + 3
end = url.index("/", start)
service = url[start : end]
start = end + 1
end = url.find(".git") if ".git" in url else len(url)
repository = url[start : end]
else:
raise RuntimeError("Invalid url: " + url)
# Esto permite utilizar un Host ficticio si tenemos 2 cuentas de bitbucket (personal y trabajo)
if service.endswith(".bitbucket.org"):
service = "bitbucket.org"
return [service, repository]
def get_branch_env_and_type(branch: str) -> tuple[str, str]:
components = branch.split("/")
target_branches = Git.flow_config("flow.branches").firstline().split(" ")
if len(components) == 2:
if components[0] not in BRANCH_TYPES:
raise ValueError(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 ValueError(
"Branch release inválido, debe tener el siguiente formato: "
"release/<env>/<type>/<name>, pero es: " + branch
)
return (components[1], components[2])
else:
raise ValueError("Branch inválido: " + branch)
def tag_command():
ensure_initialized()
ci = len(sys.argv) == 3 # ./git-flow tag <BEARER>
token = sys.argv[2] if ci else None
if token is None:
with open(REPOSITORY_TOKEN_PATH, "r") as file:
token = file.readline().strip()
output = Git("log", first_parent=True, merges=True, max_count=1, format="%H,%s").lines()
if not output:
print_error("No hay merges a taggear")
return
output = output[0]
pos = output.index(',')
commit = output[:pos]
message = output[pos+1:]
merged_branch = None
output = Git("describe", commit, tags=True, exact_match=True).without_checking().firstline()
if output:
print_error("El ultimo merge ya tiene tag: " + output)
return
if message.startswith("Merged in "):
merged_branch = message.removeprefix("Merged in ")
merged_branch = merged_branch[:merged_branch.index(' ')]
elif message.startswith("Merge branch '"):
merged_branch = message.removeprefix("Merge branch '")
merged_branch = merged_branch[:merged_branch.index("'")]
else:
print_error("No se pudo obtener el nombre de la rama mergeada")
return
remote = Git.flow_config("flow.remote").firstline()
branch = Git.get_current_branch()
output = Git("describe", abbrev="0", tags=True).without_checking().firstline()
last_version = output if output else "v1.0.0"
from_major = 1
until_dash = last_version.index("-") if "-" in last_version else None
[major, minor, patch] = map(int, last_version[from_major:until_dash].split("."))
(_, merged_branch_type) = get_branch_env_and_type(merged_branch)
increment = BRANCH_TYPES_INCREMENT[merged_branch_type]
if increment == "major":
major = major + 1
minor = 0
patch = 0
elif increment == "minor":
minor = minor + 1
patch = 0
elif increment == "patch":
patch = patch + 1
new_version = f"v{major}.{minor}.{patch}-{branch}"
if ci:
url = Git("remote", "get-url", remote).firstline()
if url:
[service, repository] = parse_url(url)
create_tag(service, repository, token, new_version, commit)
else:
Git("tag", new_version, commit).exec()
def release_command():
ensure_initialized()
branch = ensure_right_branch()
env_branches = Git.flow_config("flow.branches").firstline().split(" ")
(env, _) = get_branch_env_and_type(branch)
if env == env_branches[-1]:
raise RuntimeError("No se puede hacer release de una rama en el ultimo entorno.")
next_env = env_branches[env_branches.index(env) + 1]
boundary_commits = list(filter(
lambda c: c.startswith("-"), # boundary commits
Git("rev-list", env + "..." + branch, topo_order=True, boundary=True).lines()
))
if not boundary_commits:
raise RuntimeError("No se encontro un commit base para realizar el rebase.")
base = boundary_commits[-1].removeprefix("-")
next_branch = branch.replace(f"/{env}/", f"/{next_env}/") if branch.startswith("release/") else f"release/{next_env}/{branch}"
Git("switch", next_branch, create=True).exec()
commits = int(Git("rev-list", base + ".." + next_branch, count=True).firstline())
if commits > 1 and confirm(f"Desea reemplazar los {commits} commits de la rama por uno solo?"):
Git("reset", base, soft=True).exec()
commit_type = choose("Tipo de commit", COMMIT_TYPES[:-1]) # Solo permitir commits NO wip
Git("commit", m=commit_type + ": " + get_commit_message()).exec()
Git("rebase", base, next_branch, onto=next_env).exec()
def help_command(name: str|None = None):
if name is not None:
print_error(f"Comando inválido: {name}")
script = sys.argv[0]
print(f"""USO:
{script} <COMANDO>
COMANDOS:
init inicializa el repositorio por única vez para utilizar git-flow
new crea una nueva rama para realizar cambios
commit crea un commit siguiendo el formato de Conventional Commit
merge actualiza el changelog y crea un nuevo PR para mergear la rama
tag crea un nuevo tag para el ultimo merge
release crea una nueva rama release para pasar la rama actual al siguiente entorno
help muestra este texto de ayuda""", file=sys.stderr)
def ensure_initialized():
initialized = Git.flow_config("flow.initialized").firstline()
if initialized == "":
raise RuntimeError("El repositorio no fue inicializado, antes de continuar ejecute `git flow init`.")
elif initialized != "true":
raise RuntimeError(f"El valor de `flow.initialized` ({initialized}) es inválido.")
def get_commit_message():
message = None
while message is None:
message = input("Mensaje (recomendado: 100 caracteres): ").strip()
if not message:
print_error("El mensaje no puede ser vacío.")
message = None
return message
def ensure_right_branch():
branch = Git.get_current_branch()
if branch != "HEAD":
if confirm(f"Se encuentra sobre la rama '{branch}'. ¿Es correcto?"):
return branch
raise RuntimeError("Abortando operación")
def get_base_branch(branch: str):
if branch.startswith("release/"):
start = len("release/")
end = branch.index("/", start)
return branch[start:end]
else:
return Git.flow_config("flow.branches").firstline().split(" ")[0]
def get_unique_branch_name(branch_type):
print_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 = input("Palabras clave (por ej. 'edit worker form'): ")
if len(keywords) == 0:
print_error("Debe ingresar al menos una palabra clave.")
else:
keywords = list(filter(lambda k: len(k) > 0, keywords.split(" ")))
branch = branch_type + "/" + "-".join(keywords)
if branch in branches:
print_error(
f"La rama '{branch}' ya existe, cambie las palabras claves para generar otra"
)
branch = None
return branch
def update_changelog(branch: str, target: str):
last_commit = Git("show", "HEAD", no_patch=True, format="%s").firstline()
if last_commit == Git.UPDATED_CHANGELOG_MSG:
return False
else:
content = generate_changelog_entry(branch, target)
if os.path.isfile(Git.CHANGELOG_FILE):
with open(Git.CHANGELOG_FILE, "r") as changelog:
content = content + changelog.read()
with open(Git.CHANGELOG_FILE, "w") as changelog:
changelog.write(content)
return True
def get_changelog_header_lines(branch: str):
name = Git("config", "user.name").firstline()
email = Git("config", "user.email").firstline()
now = datetime.datetime.now()
weekday = now.strftime("%A").capitalize()
month = now.strftime("%B").capitalize()
timestamp = now.strftime(f"{weekday}, %e de {month} de %Y, %H:%M")
return [
f"## {timestamp}",
"",
f"- Autor: [{name}](mailto:{email})",
f"- Rama: `{branch}`",
"",
"### Commits",
""
]
def get_changelog_content_lines(target: str):
lines = []
commits = Git("log", target + "..", first_parent=True, format="%h", merges=False).lines()
for commit in commits:
message = Git("show", commit, no_patch=True, format="%s").firstline()
if message == Git.UPDATED_CHANGELOG_MSG:
continue
index = message.index(":")
commit_type = message[:index]
commit_message = message[index+1:]
lines.append(f"- **{commit_type}**: {commit_message} ({commit})")
diff_tree = Git("diff-tree", commit, r=True, commit_id=False, name_status=True).lines()
for diff in diff_tree:
[status, filename] = diff.split("\t")
lines.append(f"\t- {filename} ({status})")
return lines
def get_changelog_footer_lines():
return [
"",
"---",
"",
""
]
def generate_changelog_entry(branch: str, target: str):
entry_lines = get_changelog_header_lines(branch)
entry_lines += get_changelog_content_lines(target)
entry_lines += get_changelog_footer_lines()
return "\n".join(entry_lines)
def create_pull_request(service: str, repository: str, source: str, destination: str):
endpoint = get_api_endpoint(service, repository, "/pullrequests")
title = source.replace("/", ": ").replace("-", " ")
description = "\n".join(get_changelog_content_lines(destination))
headers = get_headers()
json = {
"title": title,
"description": description,
"source": {"branch": {"name": source}},
"destination": {"branch": {"name": destination}},
"close_source_branch": True
}
request = requests.post(endpoint, headers=headers, json=json)
if request.ok:
json = request.json()
print_success("PR creado exitosamente: " + json['links']['html']['href'])
else:
print_error("Ocurrió un error al crear el PR: " + request.text)
def create_tag(service: str, repository: str, token: str, tag: str, commit: str):
endpoint = get_api_endpoint(service, repository, "/refs/tags")
headers = get_headers(token)
json = {
"name": tag,
"target": {
"hash": commit
}
}
response = requests.post(endpoint, headers=headers, json=json)
if response.ok:
print("Tag creado exitosamente.")
else:
print_error("Ocurrió un error al crear el tag: " + response.text)
def get_headers(token: str | None = None):
if not token:
with open(REPOSITORY_TOKEN_PATH) as file:
token = file.readline().strip()
return {
"Accept": "application/json",
"Content-Type": "application/json",
"Authorization": "Bearer " + token,
}
def get_api_endpoint(service: str, repository: str, path: str):
if service == "bitbucket.org":
return "https://api.bitbucket.org/2.0/repositories/" + repository + path
return ""