243 lines
7.7 KiB
Python
243 lines
7.7 KiB
Python
import subprocess
|
||
from dataclasses import dataclass
|
||
|
||
from rich.markup import escape
|
||
from rich.panel import Panel
|
||
from rich.text import Text
|
||
|
||
from git_flow import FLOWCONFIG_FILENAME
|
||
from git_flow.io import *
|
||
|
||
|
||
@dataclass
|
||
class Commit:
|
||
"""Datos de un commit necesarios para versionar y generar el changelog."""
|
||
|
||
sha: str
|
||
subject: str
|
||
email: str
|
||
is_merge: bool = False
|
||
|
||
|
||
class Git:
|
||
class Status:
|
||
worktree: str
|
||
index: str
|
||
file: str
|
||
|
||
def __init__(self, line: str) -> None:
|
||
# Formato porcelain "XY <file>": X es el estado en el indice, Y en el worktree
|
||
self.index = line[0]
|
||
self.worktree = 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 is_merging() -> bool:
|
||
"""Indica si hay un merge en curso (existe MERGE_HEAD), con o sin conflictos."""
|
||
return Git("rev-parse", "MERGE_HEAD", quiet=True, verify=True).code() == 0
|
||
|
||
@staticmethod
|
||
def status():
|
||
return list(map(Git.Status, Git("status", porcelain=True).lines(False)))
|
||
|
||
@staticmethod
|
||
def get_current_tag(suffix: str = ""):
|
||
"""Último tag SemVer alcanzable desde HEAD, con el sufijo exacto indicado."""
|
||
return Git(
|
||
"describe",
|
||
abbrev="0",
|
||
tags=True,
|
||
match="v[0-9]*.[0-9]*.[0-9]*" + suffix,
|
||
exclude=None if suffix else "*-*",
|
||
).firstline(check=False, print="Obteniendo tag actual")
|
||
|
||
@staticmethod
|
||
def get_promoted_tag(merge_commit: str, suffix: str) -> str | None:
|
||
"""Si el merge trae tags con el sufijo indicado que la rama destino no tenía (por ejemplo,
|
||
un merge de dev a test trae tags *-dev nuevos), devuelve el más alto de ellos."""
|
||
tags = Git(
|
||
"tag",
|
||
"v[0-9]*.[0-9]*.[0-9]*" + suffix,
|
||
list=True,
|
||
merged=merge_commit + "^2",
|
||
no_merged=merge_commit + "^1",
|
||
sort="-v:refname",
|
||
).lines()
|
||
|
||
return tags[0] if tags else None
|
||
|
||
@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_commits(base: str, branch: str, first_parent: bool = True) -> list[Commit]:
|
||
"""Commits propios de la rama (sin merges), del más nuevo al más viejo."""
|
||
lines = Git(
|
||
"log",
|
||
base + ".." + branch,
|
||
first_parent=first_parent or None,
|
||
merges=False,
|
||
format="%h%x09%ae%x09%s",
|
||
).lines()
|
||
commits = []
|
||
|
||
for line in lines:
|
||
[sha, email, subject] = line.split("\t", 2)
|
||
commits.append(Commit(sha, subject, email))
|
||
|
||
return commits
|
||
|
||
@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 [l.strip() for l in 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, check=False)
|
||
|
||
# Sin --verbose solo se muestra el paso; si falla, la salida la muestra main()
|
||
if print is not None and settings.verbose:
|
||
self._print_process(process, print)
|
||
elif print is not None:
|
||
console.print(Text.assemble(("› ", "cyan"), 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, process: subprocess.CompletedProcess, title: str | None):
|
||
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)
|
||
|
||
ok = process.returncode == 0
|
||
# git escribe mensajes informativos en stderr, así que solo se resalta si falló
|
||
stderr_style = "dim" if ok else "red"
|
||
text = Text.assemble(("$ ", "dim"), (" ".join(command), "bold yellow"))
|
||
|
||
for line in process.stdout.splitlines():
|
||
text.append("\n" + line)
|
||
|
||
for line in process.stderr.splitlines():
|
||
text.append("\n" + line, stderr_style)
|
||
|
||
console.print(
|
||
Panel(
|
||
text,
|
||
title=f"[bold]{escape(title or 'Ejecutando')}[/bold]",
|
||
title_align="left",
|
||
subtitle=None if ok else f"código de salida {process.returncode}",
|
||
subtitle_align="right",
|
||
border_style="bright_black" if ok else "red",
|
||
expand=False,
|
||
)
|
||
)
|