feature: add branch clean subcommand

This commit is contained in:
jt
2026-09-22 01:32:10 -03:00
parent 5ece118c05
commit 6a4f6c9804
4 changed files with 125 additions and 1 deletions
+118
View File
@@ -0,0 +1,118 @@
import csv
from dataclasses import dataclass
from typing import Annotated
import typer
from git_flow import GitFlowError
from git_flow.command import base
from git_flow.git import Git
REPORT_FORMAT = "%(refname:short)\t%(objectname)\t%(authorname)\t%(authordate:iso-strict)"
REPORT_FILENAME = "branch-clean-report.csv"
app = typer.Typer()
@dataclass
class BranchInfo:
name: str
sha: str
author: str
date: str
@app.command()
def clean(
remote: Annotated[
bool,
typer.Option(
help="Elimina las ramas remotas alcanzables desde HEAD, en vez de las locales"
),
] = False,
):
"""Elimina las ramas alcanzables desde HEAD, dejando un reporte CSV de las ramas eliminadas"""
base.ensure_initialized()
branches = get_remote_branches() if remote else get_local_branches()
if not branches:
base.io_info("No hay ramas alcanzables desde HEAD para eliminar.")
return
base.io_info(
"\n".join(f"- {branch.name}" for branch in branches),
title=f"Ramas a eliminar ({'remoto' if remote else 'local'})",
)
if not base.io_confirm(f"¿Eliminar {len(branches)} rama(s)?", False):
raise GitFlowError("Ejecución abortada")
write_report(branches)
for branch in branches:
delete_branch(branch, remote)
base.io_success(f"Se eliminaron {len(branches)} rama(s). Reporte generado en {REPORT_FILENAME}")
def get_local_branches() -> list[BranchInfo]:
current_branch = Git.get_current_branch()
branches = get_merged_branches("refs/heads/")
return [b for b in branches if b.name != current_branch and b.name not in base.environments]
def get_remote_branches() -> list[BranchInfo]:
if "flow.remote" not in base.flowconfig:
raise GitFlowError("El repositorio no tiene configurado un remoto.")
remote = base.flowconfig["flow.remote"]
current_tracking = Git.get_tracking_branch(Git.get_current_branch())
branches = get_merged_branches(f"refs/remotes/{remote}/")
result = []
for branch in branches:
if branch.name == current_tracking or branch.name == f"{remote}/HEAD":
continue
name = branch.name.removeprefix(f"{remote}/")
if name in base.environments:
continue
result.append(BranchInfo(name, branch.sha, branch.author, branch.date))
return result
def get_merged_branches(pattern: str) -> list[BranchInfo]:
lines = Git("for-each-ref", pattern, merged="HEAD", format=REPORT_FORMAT).lines()
return [parse_branch(line) for line in lines]
def parse_branch(line: str) -> BranchInfo:
name, sha, author, date = line.split("\t")
return BranchInfo(name, sha, author, date)
def write_report(branches: list[BranchInfo]):
with open(REPORT_FILENAME, "w", newline="") as f:
writer = csv.writer(f)
writer.writerow(["name", "commit sha", "author", "date"])
for branch in branches:
writer.writerow([branch.name, branch.sha, branch.author, branch.date])
def delete_branch(branch: BranchInfo, remote: bool):
if remote:
remote_name = base.flowconfig["flow.remote"]
Git("push", remote_name, branch.name, delete=True).exec(
print=f"Eliminando rama remota '{branch.name}'"
)
else:
Git("branch", branch.name, d=True).exec(
print=f"Eliminando rama local '{branch.name}'"
)