166 lines
5.0 KiB
Python
166 lines
5.0 KiB
Python
import subprocess
|
|
|
|
from git_flow import FLOWCONFIG_FILENAME
|
|
from git_flow.io import *
|
|
|
|
|
|
class Git:
|
|
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")
|
|
|
|
command: list[str]
|
|
|
|
@staticmethod
|
|
def get_config(file: str | None = None):
|
|
config: dict[str, str] = {}
|
|
lines = Git("config", file=file, list=True).lines()
|
|
|
|
for line in lines:
|
|
pos = line.index("=")
|
|
key = line[:pos]
|
|
value = line[pos + 1 :]
|
|
config[key] = value
|
|
|
|
return config
|
|
|
|
@staticmethod
|
|
def set_config(config: dict[str, str], file: str | None = None):
|
|
for key, value in config.items():
|
|
Git("config", key, value, file=file).exec()
|
|
|
|
@staticmethod
|
|
def flow_config(*args, **kwargs):
|
|
return Git("config", *args, file=FLOWCONFIG_FILENAME, **kwargs)
|
|
|
|
@staticmethod
|
|
def get_current_branch():
|
|
return Git("rev-parse", "HEAD", abbrev_ref=True).firstline()
|
|
|
|
@staticmethod
|
|
def get_branches(prefix: str = ""):
|
|
return Git._get_references("heads/" + prefix)
|
|
|
|
@staticmethod
|
|
def status():
|
|
return list(map(Git.Status, Git("status", porcelain=True).lines(False)))
|
|
|
|
@staticmethod
|
|
def get_current_tag():
|
|
return Git("describe", abbrev="0", tags=True).firstline(check=False, print="Obteniendo tag actual")
|
|
|
|
@staticmethod
|
|
def get_first_fork_point(branch: str, env: str):
|
|
# Lista de commits alcanzables por el entorno (por ejemplo, dev),
|
|
env_commits = Git("rev-list", env + "..." + branch, first_parent=True).lines()
|
|
|
|
if not env_commits:
|
|
raise RuntimeError("No se encontro un commit base para realizar el rebase.")
|
|
|
|
return Git("show", env_commits[-1] + "^", patch=False, format="%h").firstline()
|
|
|
|
@staticmethod
|
|
def _get_references(kind: str):
|
|
prefix = "refs/" + kind
|
|
return Git("for-each-ref", prefix + "**", format="%(refname:short)").lines()
|
|
|
|
@staticmethod
|
|
def is_repository() -> bool:
|
|
return Git("rev-parse", is_inside_work_tree=True).code() == 0
|
|
|
|
@staticmethod
|
|
def get_tracking_branch(branch: str) -> str | None:
|
|
try:
|
|
return Git("rev-parse", branch + "@{upstream}", abbrev_ref=True).firstline()
|
|
except subprocess.CalledProcessError:
|
|
return None
|
|
|
|
def __init__(
|
|
self, subcommand: str, *args: str, **kwargs: str | int | bool | list[str] | None
|
|
) -> 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)
|
|
elif value is None:
|
|
continue
|
|
elif isinstance(value, list):
|
|
for v in value:
|
|
self.command.append("--" + option + "=" + v)
|
|
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, **kwargs):
|
|
process = self._get(**kwargs)
|
|
lines = process.stdout.splitlines()
|
|
|
|
return lines if not strip else list(map(lambda l: l.strip(), lines))
|
|
|
|
def firstline(self, **kwargs):
|
|
lines = self.lines(**kwargs)
|
|
|
|
return lines[0] if lines else ""
|
|
|
|
def code(self, **kwargs):
|
|
return self._run(check=False, **kwargs).returncode
|
|
|
|
def exec(self, **kwargs):
|
|
self._run(**kwargs)
|
|
|
|
def _get(self, print: str | None = None, check: bool = True):
|
|
process = subprocess.run(self.command, capture_output=True, text=True)
|
|
|
|
if print is not None:
|
|
self._print_process(process.stdout, process.stderr, print)
|
|
|
|
if check:
|
|
process.check_returncode()
|
|
|
|
return process
|
|
|
|
def _run(self, print: str | None = None, check: bool = True):
|
|
return self._get(print, check)
|
|
|
|
def _print_process(self, stdout: str, stderr: str, title: 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)
|
|
|
|
text = f"[green]$ {" ".join(command)}[/green]"
|
|
|
|
for line in stdout.splitlines():
|
|
text += "\n" + line
|
|
|
|
for line in stderr.splitlines():
|
|
text += "\n[red]" + line + "[/red]"
|
|
|
|
panel(title or "Ejecutando", text)
|