refactor: clase Git para ejecutar comandos
This commit is contained in:
@@ -0,0 +1,2 @@
|
||||
.repository-token
|
||||
**/__pycache__/
|
||||
@@ -0,0 +1,3 @@
|
||||
# Git Flow
|
||||
|
||||
Herramienta para establecer un workflow y versionar un repositorio
|
||||
@@ -0,0 +1,577 @@
|
||||
#!./venv/bin/python3
|
||||
|
||||
import os
|
||||
from subprocess import CalledProcessError
|
||||
import sys
|
||||
import datetime
|
||||
import requests
|
||||
|
||||
import lib.git as git
|
||||
from lib.git import Git
|
||||
from lib.io import *
|
||||
|
||||
|
||||
REPOSITORY_TOKEN_PATH=".repository-token"
|
||||
|
||||
|
||||
def main():
|
||||
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 == "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_base_branch(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)
|
||||
|
||||
return [service, repository]
|
||||
|
||||
|
||||
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).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).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 = merged_branch[:merged_branch.index('/')] if '/' in merged_branch else 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 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} [help | init | new | commit]
|
||||
|
||||
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
|
||||
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 generate_changelog_entry(branch: str, target: str):
|
||||
name = Git("config", "user.name").firstline()
|
||||
email = Git("config", "user.email").firstline()
|
||||
timestamp = datetime.datetime.now().strftime("%d/%m/%Y %H:%M")
|
||||
|
||||
entry_lines = [
|
||||
f"## {name} ({email}) - {timestamp}",
|
||||
"",
|
||||
f"**{branch}**",
|
||||
]
|
||||
|
||||
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:]
|
||||
|
||||
entry_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")
|
||||
entry_lines.append(f"\t- {filename} ({status})")
|
||||
|
||||
entry_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("-", " ")
|
||||
lines = generate_changelog_entry(source, destination).splitlines()
|
||||
description = "\n".join(lines[3:-4])
|
||||
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:
|
||||
print("PR creado exitosamente.")
|
||||
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 ""
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+251
@@ -0,0 +1,251 @@
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
from lib.io import *
|
||||
|
||||
FLOWCONFIG_FILE = ".flowconfig"
|
||||
CHANGELOG_FILE = "CHANGELOG.md"
|
||||
UPDATED_CHANGELOG_MSG = "Updated " + CHANGELOG_FILE
|
||||
|
||||
|
||||
def branch_exists(branch):
|
||||
return os.path.isfile(os.path.join(".git", "refs", "heads", branch))
|
||||
|
||||
|
||||
def get_current_branch():
|
||||
output = os.popen("git rev-parse --abbrev-ref HEAD").readline()
|
||||
output = output.strip()
|
||||
|
||||
return None if output == "HEAD" else output
|
||||
|
||||
|
||||
def is_repository():
|
||||
return os.path.isdir(".git")
|
||||
|
||||
|
||||
def flow_config(name: str, value: str | None = None) -> str | None:
|
||||
if value is None:
|
||||
value = os.popen(f"git config --file {FLOWCONFIG_FILE} '{name}'").readline()
|
||||
return value.strip()
|
||||
else:
|
||||
os.popen(f"git config --file {FLOWCONFIG_FILE} '{name}' '{value}'").close()
|
||||
|
||||
|
||||
def config(name: str, value: str | None = None, **kwargs) -> str | None:
|
||||
options = []
|
||||
|
||||
for k in kwargs:
|
||||
value = kwargs[k]
|
||||
k = k.replace("_", "-")
|
||||
|
||||
if isinstance(value, bool):
|
||||
options.append("--" + k if value else "--no-" + k)
|
||||
else:
|
||||
options.append(f"--{k}={value}")
|
||||
|
||||
options = " ".join(options)
|
||||
|
||||
if value is None:
|
||||
return os.popen(f"git config {options} '{name}'").readline().strip()
|
||||
else:
|
||||
os.popen(f"git config {options} '{name}' '{value}'").close()
|
||||
|
||||
|
||||
def create_branch(name, base: str | None = None, switch: bool = False):
|
||||
command = "switch -c" if switch else "branch"
|
||||
os.popen(f"git {command} '{name}'" + (f" '{base}'" if base is not None else ""))
|
||||
|
||||
|
||||
def get_remotes():
|
||||
return os.popen("git remote").readlines()
|
||||
|
||||
|
||||
class Status:
|
||||
worktree: str
|
||||
index: str
|
||||
file: str
|
||||
|
||||
def __init__(self, line: str) -> None:
|
||||
self.worktree = line[0]
|
||||
self.index = line[1]
|
||||
self.file = line[3:].strip("\n")
|
||||
|
||||
|
||||
def status():
|
||||
status: list[Status] = []
|
||||
lines = os.popen("git status --porcelain").readlines()
|
||||
|
||||
for line in lines:
|
||||
status.append(Status(line))
|
||||
|
||||
return status
|
||||
|
||||
|
||||
def switch(branch: str = "-"):
|
||||
os.popen(f"git switch '{branch}'").close()
|
||||
|
||||
|
||||
def rename_branch(oldbranch: str, newbranch):
|
||||
os.popen(f"git branch -m '{oldbranch}' '{newbranch}'").close()
|
||||
|
||||
|
||||
def _command_list_as_str(cmd: list[str]) -> str:
|
||||
s = []
|
||||
|
||||
for arg in cmd:
|
||||
arg = ('"' + arg + '"') if " " in arg else arg
|
||||
s.append(arg)
|
||||
|
||||
return " ".join(s)
|
||||
|
||||
|
||||
def _run_command(git_subcommand: str, *args, **kwargs) -> subprocess.CompletedProcess:
|
||||
cmd = ["git", git_subcommand]
|
||||
|
||||
for k in kwargs:
|
||||
value = kwargs[k]
|
||||
k = k.replace("_", "-")
|
||||
|
||||
if isinstance(value, bool):
|
||||
cmd.append("--" + k if value else "--no-" + k)
|
||||
else:
|
||||
if len(k) == 1:
|
||||
cmd.append("-" + k)
|
||||
else:
|
||||
cmd.append("--" + k)
|
||||
|
||||
cmd.append(value)
|
||||
|
||||
cmd += args
|
||||
|
||||
print(f"> Running: " + _command_list_as_str(cmd))
|
||||
|
||||
return subprocess.run(cmd, capture_output=True, text=True)
|
||||
|
||||
|
||||
def output(git_subcommand: str, *args, **kwargs) -> list[str] | None:
|
||||
completed_process = _run_command(git_subcommand, *args, **kwargs)
|
||||
|
||||
if completed_process.returncode == 0:
|
||||
return list(filter(lambda l: l, str(completed_process.stdout).splitlines()))
|
||||
else:
|
||||
for line in str(completed_process.stderr).splitlines():
|
||||
print("! " + line, file=sys.stderr)
|
||||
|
||||
|
||||
def exec(git_subcommand: str, *args, **kwargs) -> bool:
|
||||
return _run_command(git_subcommand, *args, **kwargs).returncode == 0
|
||||
|
||||
class Git:
|
||||
FLOWCONFIG_FILE = ".flowconfig"
|
||||
|
||||
command: list[str]
|
||||
check_returncode: bool = True
|
||||
|
||||
@staticmethod
|
||||
def flow_config(*args, **kwargs):
|
||||
return Git("config", *args, file=FLOWCONFIG_FILE, **kwargs)
|
||||
|
||||
@staticmethod
|
||||
def get_current_branch():
|
||||
return Git("rev-parse", "HEAD", abbrev_ref=True).firstline()
|
||||
|
||||
@staticmethod
|
||||
def get_branches():
|
||||
return Git._get_references("heads")
|
||||
|
||||
@staticmethod
|
||||
def status():
|
||||
return list(map(Status, Git("status", porcelain=True).lines(False)))
|
||||
|
||||
@staticmethod
|
||||
def _get_references(kind: str):
|
||||
prefix = "refs/" + kind + "/"
|
||||
return map(
|
||||
lambda r: r.removeprefix(prefix),
|
||||
filter(
|
||||
lambda l: l.startswith(prefix),
|
||||
Git("for-each-ref", format="%(refname)").lines()
|
||||
)
|
||||
)
|
||||
|
||||
def __init__(self, subcommand: str, *args: str, **kwargs: str | int | bool) -> None:
|
||||
self.command = ["git", subcommand]
|
||||
|
||||
for option, value in kwargs.items():
|
||||
option = option.replace("_", "-")
|
||||
|
||||
if isinstance(value, bool):
|
||||
if len(option) == 1:
|
||||
self.command.append("-" + option)
|
||||
else:
|
||||
prefix = "--" if value else "--no-"
|
||||
self.command.append(prefix + option)
|
||||
else:
|
||||
value = value if isinstance(value, str) else str(value)
|
||||
|
||||
if len(option) == 1:
|
||||
self.command.append("-" + option)
|
||||
self.command.append(value)
|
||||
else:
|
||||
self.command.append("--" + option + '=' + value)
|
||||
|
||||
self.command += args
|
||||
|
||||
def lines(self, strip: bool = True):
|
||||
process = subprocess.run(self.command, capture_output=True, text=True)
|
||||
|
||||
self.__print_process(process.stdout, process.stderr)
|
||||
|
||||
if self.check_returncode:
|
||||
process.check_returncode()
|
||||
|
||||
lines = process.stdout.splitlines()
|
||||
|
||||
return lines if not strip else list(map(lambda l: l.strip(), lines))
|
||||
|
||||
def firstline(self):
|
||||
lines = self.lines()
|
||||
|
||||
return lines[0] if lines else ""
|
||||
|
||||
def exec(self):
|
||||
process = subprocess.run(self.command, capture_output=True, text=True)
|
||||
|
||||
self.__print_process(process.stdout, process.stderr)
|
||||
|
||||
if self.check_returncode:
|
||||
process.check_returncode()
|
||||
|
||||
def code(self):
|
||||
process = subprocess.run(self.command, capture_output=True, text=True)
|
||||
|
||||
self.__print_process(process.stdout, process.stderr)
|
||||
|
||||
return process.returncode
|
||||
|
||||
def without_checking(self):
|
||||
self.check_returncode = False
|
||||
|
||||
return self
|
||||
|
||||
def __print_process(self, stdout: str, stderr: str):
|
||||
command = []
|
||||
|
||||
for arg in self.command:
|
||||
if arg.startswith("--") and "=" in arg:
|
||||
pos = arg.index("=")
|
||||
arg = arg[:pos+1] + '"' + arg[pos+1:] + '"'
|
||||
elif " " in arg:
|
||||
arg = '"' + arg + '"'
|
||||
|
||||
command.append(arg)
|
||||
|
||||
print(COLOR_YELLOW + "> " + " ".join(command) + COLOR_RESET)
|
||||
|
||||
for line in stdout.splitlines():
|
||||
print(COLOR_BLUE + "< " + line + COLOR_RESET)
|
||||
|
||||
for line in stderr.splitlines():
|
||||
print(COLOR_RED + "! " + line + COLOR_RESET)
|
||||
@@ -0,0 +1,61 @@
|
||||
import sys
|
||||
|
||||
|
||||
COLOR_BLACK = '\033[30m'
|
||||
COLOR_RED = '\033[31m'
|
||||
COLOR_GREEN = '\033[32m'
|
||||
COLOR_YELLOW = '\033[33m'
|
||||
COLOR_BLUE = '\033[34m'
|
||||
COLOR_BLACK_BOLD = '\033[1;30m'
|
||||
COLOR_RESET = '\033[0m'
|
||||
|
||||
|
||||
def confirm(prompt, default: bool = True):
|
||||
user_input = input(prompt + (" [Y/n]: " if default else " [y/N]: "))
|
||||
|
||||
return default if len(user_input) == 0 else user_input.startswith("y")
|
||||
|
||||
|
||||
def choose(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:
|
||||
user_input = input(f"Seleccione una opción [1-{len(options)}] o escribala: ")
|
||||
|
||||
if user_input.isdigit():
|
||||
user_input = int(user_input)
|
||||
|
||||
if 1 <= user_input and user_input <= len(options):
|
||||
selection = options[user_input - 1]
|
||||
|
||||
if not confirm(
|
||||
f"Seleccionó la opción {user_input} ({selection}), ¿es correcto?"
|
||||
):
|
||||
selection = None
|
||||
else:
|
||||
print_error(f"La opción {user_input} está fuera del rango permitido.")
|
||||
elif user_input in options:
|
||||
selection = user_input
|
||||
else:
|
||||
print_error(f"La opción '{user_input}' es inválida.")
|
||||
|
||||
return selection
|
||||
|
||||
|
||||
def print_error(message: str):
|
||||
print(f"{COLOR_RED}[err] {message}{COLOR_RESET}", file=sys.stderr)
|
||||
|
||||
|
||||
def print_warning(message: str):
|
||||
print(f"{COLOR_YELLOW}[wrn] {message}{COLOR_RESET}")
|
||||
|
||||
def print_info(message: str):
|
||||
print(f"{COLOR_BLUE}[inf] {message}{COLOR_RESET}")
|
||||
|
||||
def print_debug(message: str):
|
||||
print(f"{COLOR_BLACK_BOLD}[dbg] {message}{COLOR_RESET}")
|
||||
@@ -0,0 +1,4 @@
|
||||
{
|
||||
"$schema": "https://raw.githubusercontent.com/microsoft/pyright/main/packages/vscode-pyright/schemas/pyrightconfig.schema.json",
|
||||
"pythonVersion": "3.12"
|
||||
}
|
||||
Reference in New Issue
Block a user