refactor: clase Git para ejecutar comandos
This commit is contained in:
+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}")
|
||||
Reference in New Issue
Block a user