""" Utilitários para gerenciar versões de projetos, incluindo leitura de changelogs, detecção de tipos de release e atualização de arquivos de configuração. Este módulo fornece funções para determinar a versão atual do projeto com base em tags Git e no conteúdo do arquivo CHANGELOG.md, identificar o tipo de release (major, minor, patch) e atualizar arquivos de configuração, como template.yaml, com a nova versão. As funções são projetadas para serem usadas em scripts de CI/CD e automação de tarefas relacionadas ao gerenciamento de versões, garantindo que as versões sejam gerenciadas de forma consistente e que os arquivos de configuração sejam atualizados corretamente. """ from __future__ import annotations import os import re from pathlib import Path from common import print_message from src.datalake.commons.ci_utils import run_cmd def _read_text(path: Path) -> str: return path.read_text(encoding="utf-8") def _latest_tag_version() -> str: run_cmd(["git", "fetch", "--tags"]) rev = run_cmd( ["git", "rev-list", "--tags", "--max-count=1"], capture_output=True, check=False ) rev_sha = (rev.stdout or "").strip() raw = "" if rev_sha: result = run_cmd( ["git", "describe", "--tags", rev_sha], capture_output=True, check=False ) raw = (result.stdout or "").strip() if not raw: result = run_cmd(["git", "tag", "--sort=-creatordate"], capture_output=True) tags = [t.strip() for t in result.stdout.splitlines() if t.strip()] raw = tags[0] if tags else "v0-0-0" match = re.search(r"^v(\d+)-(\d+)-(\d+)$", raw) if not match: return "0.0.0" return f"{match.group(1)}.{match.group(2)}.{match.group(3)}" def _branch_tag_version(ci_commit_ref_name: str) -> str: match = re.search(r"^v(\d+)-(\d+)-(\d+)$", ci_commit_ref_name or "") if not match: return "" return f"{match.group(1)}.{match.group(2)}.{match.group(3)}" def _detect_release_kind(changelog_text: str) -> str: for line in changelog_text.splitlines(): line_fixed = line.strip() if line_fixed.startswith("### "): return line_fixed return "" def get_version() -> str: target_file = Path("CHANGELOG.md") if not target_file.exists(): raise RuntimeError("[Dados] CHANGELOG.md não encontrado") content = target_file.read_bytes() if b"\r\n" in content: raise RuntimeError( "[Dados] O arquivo CHANGELOG.md esta como CRLF, corrija e tente novamente, bloqueando fluxo..." ) changelog_text = content.decode("utf-8", errors="ignore") if "TEMPLATE CHANGELOG (ALTERAR)" in changelog_text: raise RuntimeError( "[Dados] O arquivo CHANGELOG.md contem 'TEMPLATE CHANGELOG (ALTERAR)', " "remova essa linha e tente novamente, bloqueando fluxo..." ) ci_commit_ref_name = os.environ.get("CI_COMMIT_REF_NAME", "") ci_commit_branch = os.environ.get("CI_COMMIT_BRANCH", "") version = _branch_tag_version(ci_commit_ref_name) release_type = "" if not version: print_message("[Dados] Operando em branch, calculando próxima versão...") version = _latest_tag_version() major, minor, patch = [int(v) for v in version.split(".")] release_kind = _detect_release_kind(changelog_text) next_version = "0.0.0" if release_kind == "### Quebras de compatibilidade": next_version = f"{major + 1}.0.0" release_type = "major" elif release_kind in {"### Novas funcionalidades", "### Melhorias"}: next_version = f"{major}.{minor + 1}.0" release_type = "minor" elif release_kind == "### Correções": next_version = f"{major}.{minor}.{patch + 1}" release_type = "patch" if next_version == "0.0.0" and ci_commit_branch != "develop": raise RuntimeError( "[Dados] Changelog não preenchido, coloque o que foi feito e tente novamente..." ) if next_version in {".1.0", "..1"}: release_type = "major" next_version = "1.0.0" next_version = f"{next_version}-SNAPSHOT" else: print_message(f"[Dados] Operando em tag, usando versão da tag: {version}") next_version = version print_message(f"[Dados] NEXTVERSION: {next_version} - RELEASE_TYPE: {release_type}") os.environ["NEXTVERSION"] = next_version if release_type: os.environ["RELEASE_TYPE"] = release_type return next_version def check_snapshots(target_file: str) -> None: path = Path(target_file) if not path.exists(): return lines = _read_text(path).splitlines()[4:] joined = "\n".join(lines) if re.search( r"(-SNAPSHOT|_SNAPSHOT|-alpha|_alpha|-beta|_beta|-prerelease|_prerelease|-pre-release|_pre-release|_dev|\.dev)", joined, ): raise RuntimeError( f"[Dados] O arquivo {target_file} contem dependencia de versão aberta " "(SNAPSHOT, alpha, beta, prerelease, dev), bloqueando fluxo..." ) print_message( f"[Dados] O arquivo {target_file} não contem dependencia de versão aberta, seguindo..." ) def fix_template_yaml_version() -> None: next_version = get_version() target_file = Path("template.yaml") if not target_file.exists(): raise RuntimeError("[Dados] template.yaml não encontrado") lines = target_file.read_text(encoding="utf-8").splitlines() desc_idx = -1 for idx, line in enumerate(lines): if line.strip().startswith("Description:"): desc_idx = idx break if desc_idx < 0: raise RuntimeError("[Dados] Description não encontrada em template.yaml") if "Description: >" not in lines[desc_idx]: prefix = lines[desc_idx].split("Description:", 1)[1] lines[desc_idx] = "Description: >" lines.insert(desc_idx + 1, f" {prefix.strip()}") line_num = desc_idx + 1 if line_num >= len(lines): lines.append(f" - Version: {next_version}") else: if "- Version:" not in lines[line_num]: lines[line_num] = f"{lines[line_num]} - Version:" lines[line_num] = re.sub( r" - Version:.*$", f" - Version: {next_version}", lines[line_num] ) target_file.write_text("\n".join(lines) + "\n", encoding="utf-8") def release_project_version() -> None: check_snapshots("template_jobs.yaml") check_snapshots("template_jobs_consolidate.yaml") get_version() release_type = os.environ.get("RELEASE_TYPE", "") print_message( f"[Dados] Efetuando procedimento de release do devops: {release_type}" ) if release_type: run_cmd(["bash", "senior-ci/ci/ci.sh", "-r", release_type])