git-flow/git-flow

#!/usr/bin/env python3

import os
import sys
import datetime
import requests
import locale

import lib.git as git
from lib.git import Git
from lib.io import *


REPOSITORY_TOKEN_PATH=".repository-token"
LOCALE = 'es_AR.UTF-8'


def main():
    try:
        locale.setlocale(locale.LC_ALL, LOCALE)
    except locale.Error as e:
        print_warning(f"No se pudo configurar el locale '{LOCALE}': {e}")

    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)

    # Esto permite utilizar un Host ficticio si tenemos 2 cuentas de bitbucket (personal y trabajo)
    if service.endswith(".bitbucket.org"):
        service = "bitbucket.org"

    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).without_checking().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).without_checking().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("."))

    components = merged_branch.split("/")

    if len(components) not in [2, 4]:
        raise RuntimeError("Branch mergeado es inválido: " + merged_branch)
    elif len(components) == 4 and components[0] != "release":
        raise RuntimeError("Release branch mergeado es inválido: " + merged_branch)
    elif components[-2] not in BRANCH_TYPES:
        raise RuntimeError("Tipo de rama mergeado es inválido: " + components[-2])

    increment = BRANCH_TYPES_INCREMENT[components[-2]]

    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 get_changelog_header_lines(branch: str):
    name = Git("config", "user.name").firstline()
    email = Git("config", "user.email").firstline()

    now = datetime.datetime.now()
    weekday = now.strftime("%A").capitalize()
    month = now.strftime("%B").capitalize()
    timestamp = now.strftime(f"{weekday}, %e de {month} de %Y, %H:%M")

    return [
        f"## {timestamp}",
        "",
        f"- Autor: [{name}](mailto:{email})",
        f"- Rama: `{branch}`",
        "",
        "### Commits",
        ""
    ]

def get_changelog_content_lines(target: str):
    lines = []
    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:]

        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")
            lines.append(f"\t- {filename} ({status})")

    return lines

def get_changelog_footer_lines():
    return [
        "",
        "---",
        "",
        ""
    ]

def generate_changelog_entry(branch: str, target: str):
    entry_lines = get_changelog_header_lines(branch)
    entry_lines += get_changelog_content_lines(target)
    entry_lines += get_changelog_footer_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("-", " ")
    description = "\n".join(get_changelog_content_lines(destination))
    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:
        json = request.json()
        print_success("PR creado exitosamente: " + json['links']['html']['href'])
    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()