git-flow/src/git_flow/command/base.py

from git_flow import (
    BRANCH_TYPES,
    REPOSITORY_TOKEN_FILENAME,
    FLOWCONFIG_FILENAME,
    FLOWCONFIG_VERSION,
    COMMIT_TYPES,
    GitFlowError,
)
from git_flow.git import Git
from git_flow.remote.base import RemoteAPI
from git_flow.remote.bitbucket import BitbucketRemoteAPI
from git_flow.remote.github import GithubRemoteAPI
from git_flow.remote.gitea import GiteaRemoteAPI
from git_flow.io import *
from os.path import isfile

from typing import Optional
import rich
import rich.panel
import questionary

flowconfig = Git.get_config(FLOWCONFIG_FILENAME) if isfile(FLOWCONFIG_FILENAME) else {}
environments = (
    flowconfig["flow.branches"].split(",") if "flow.branches" in flowconfig else []
)


def ensure_initialized():
    initialized = flowconfig["flow.initialized"] if flowconfig else None

    if not initialized:
        raise GitFlowError(
            "El repositorio no fue inicializado, debe ejecutar el comando `init`."
        )
    elif initialized != "true":
        raise GitFlowError(
            f"El valor de `flow.initialized` ({initialized}) es inválido."
        )

    version = int(flowconfig.get("flow.version", "1"))

    if version != FLOWCONFIG_VERSION:
        raise GitFlowError(
            f"La versión del flowconfig ({version}) no es compatible con esta versión de git-flow ({FLOWCONFIG_VERSION}). "
            f"Debe reinicializar el repositorio."
        )


def ensure_right_branch():
    branch = Git.get_current_branch()

    if branch == "HEAD":
        raise GitFlowError("No se encuentra parado sobre una rama.")
    elif io_confirm(f"Rama actual: {branch}. ¿Es correcto?"):
        return branch
    else:
        raise GitFlowError("Ejecución cancelada.")


def get_branch_env_and_type(branch: str) -> tuple[str, str]:
    components = branch.split("/")

    if len(components) == 2:
        if components[0] not in BRANCH_TYPES:
            raise GitFlowError(
                f"Branch inválida, el tipo '{components[0]}' no es válido."
            )

        return (environments[0], components[0])
    elif len(components) == 4:
        valid_environments = environments[1:]

        if (
            components[0] != "release"
            or components[1] not in valid_environments
            or components[2] not in BRANCH_TYPES
        ):
            raise GitFlowError(
                "Branch release inválido, debe tener el siguiente formato: "
                "release/<env>/<type>/<name>, pero es: " + branch
            )

        return (components[1], components[2])
    else:
        raise GitFlowError("Branch inválido: " + branch)


def ensure_repository_token():
    if not isfile(REPOSITORY_TOKEN_FILENAME):
        raise GitFlowError(
            "No existe un token para el repositorio en " + REPOSITORY_TOKEN_FILENAME
        )

    with open(REPOSITORY_TOKEN_FILENAME) as f:
        return f.readline().strip()


def ensure_clean_worktree(has_remote: bool):
    status = Git.status()

    if not status:
        return

    not_empty = any(map(lambda s: s.index != " " or s.worktree != " ", status))

    if not_empty:
        if not has_remote:
            io_warning("Existen cambios en tu entorno de trabajo sin commitear.")

            if not io_confirm("¿Desea continuar?", False):
                raise GitFlowError("Ejecución abortada")
        else:
            raise GitFlowError("No se puede continuar con cambios pendientes.")


def get_remote_api(token: str) -> RemoteAPI:
    if "flow.remote" not in flowconfig:
        raise GitFlowError("El repositorio no tiene configurado un remoto.")

    remote_type = flowconfig.get("flow.remote-type", "").lower()
    [schema, host, repository] = RemoteAPI.parse(flowconfig["flow.remote"])

    if remote_type == "gitea":
        remote_schema = "https://" if schema == "ssh://" else schema
        return GiteaRemoteAPI(remote_schema + host, repository, token)
    elif remote_type == "bitbucket":
        return BitbucketRemoteAPI(repository, token)
    elif remote_type == "github":
        return GithubRemoteAPI(repository, token)
    else:
        raise GitFlowError("El tipo del remoto es inválido")


def is_valid_ticket(ticket: str) -> bool:
    """Check if the specified string is a valid ticket number (ABC-123)"""
    components = ticket.split("-")

    if len(components) != 2:
        return False

    ticket_project = components[0]
    ticket_number = components[1]

    return (
        ticket_project.isalpha()
        and ticket_project.isupper()
        and ticket_number.isnumeric()
    )


def get_commit_message(commit_types: list[str] = COMMIT_TYPES) -> str:
    commit_type = io_choice("Tipo de commit", commit_types)
    commit_message = io_prompt(
        "Mensaje de commit",
        validator=lambda s: 0 < len(s) and len(s) < 100,
        instruction="100 caracteres máximo",
    )
    return commit_type + ": " + commit_message


def io_error(message: str, title: Optional[str] = None):
    _io_status(message, "✖ Error: ", "red", title)


def io_warning(message: str, title: Optional[str] = None):
    _io_status(message, "⚠ Warning: ", "yellow", title)


def io_info(message: str, title: Optional[str] = None):
    _io_status(message, "ℹ Info: ", "blue", title)


def io_success(message: str, title: Optional[str] = None):
    _io_status(message, "✔ Success: ", "green", title)


def _io_status(message: str, prefix: str, style: str, title: Optional[str] = None):
    if title:
        rich.print(
            rich.panel.Panel(
                message,
                title=prefix + title,
                style=style,
                expand=False,
                title_align="left",
            )
        )
    else:
        rich.print(f"[{style}]{prefix}{message}[/{style}]")


def io_confirm(question: str, default: bool = True) -> bool:
    return questionary.confirm(question, default=default, auto_enter=False).unsafe_ask()


def io_choice(prompt: str, options: list[str]) -> str:
    return questionary.select(
        prompt,
        options,
        use_search_filter=True,
        use_jk_keys=False,
        instruction="Tipear o usar flechas",
    ).unsafe_ask()


def io_prompt(
    message: str,
    default: str = "",
    persistent: bool = False,
    strip: bool = False,
    validator=None,
    instruction: str | None = None,
) -> str:
    if persistent and not validator:
        validator = lambda s: len(s.strip()) > 0

    value = questionary.text(
        message, default, validate=validator, instruction=instruction
    ).unsafe_ask()

    return value.strip() if strip else value