import json import os import re import shutil from pathlib import Path from colorama import Fore from common import exec_command, exit_message, get_env_variable_required, print_message from common.file_helpers import find_file_by_name, load_strict_json from docker import push_image from dotnet import BaseDotNetCI class DotNetAppsCI(BaseDotNetCI): def exec_ci(self): for opt, arg in self.opts: if opt == "-v": self.validation() elif opt == "-r": self._release_app(arg) elif opt == "-p": self._publish_app() elif opt == "-x": self.run_sonar() elif opt == "-g": self._generate_template(arg, self.args) else: exit_message("É preciso informar a opção referente a ação escolhida") def compile(self, version): self.configure_nuget() print_message("Compilando projeto.") exec_command( f"dotnet publish --no-self-contained -c release -o output/ -p:version={version}", error_message="Compilação finalizada com erro.", ) print_message("Compilação finalizada com sucesso.", Fore.GREEN) def _release_app(self, versioning_release: str): if versioning_release == "snapshot": version = self.get_version() self._publish_app(version) else: self.release(versioning_release) def _publish_app(self, version: str = None): if not version: version = self.get_version() print_message(f"Publicando versão {version}") self.compile(version) push_image(version) def _generate_template(self, service_name=None, args=None): print_message("Gerando templates do microsserviço.") if not service_name: raise ValueError("É necessário informar o nome do serviço.") service_words = service_name.strip().split("-") service_name_formatted = ".".join(word.capitalize() for word in service_words) args = self._normalize_args(args) self.configure_nuget() scaffold_dir = "/tmp/dotnet-flex-scaffold" exec_command( f"git clone https://gitlab-ci-token:$CI_JOB_TOKEN@git.senior.com.br/devops/dotnet-flex-scaffold.git {scaffold_dir}" ) exec_command(f"dotnet new install {scaffold_dir}") repo_root = os.getcwd() exec_command( f""" dotnet new senior-dotnet-micro \ -n {service_name_formatted} \ -o . \ --EntityName {args['entityName']} \ --EntityPluralName {args['entityPluralName']} \ --SecondaryName {args['secondaryName']} \ --SecondaryPluralName {args['secondaryPluralName']} \ --Domain {args['domain']} \ --Service {args['service']} \ --SchemaPrefix {args['schemaPrefix']} """ ) main_project_dir = os.path.join(repo_root, service_name_formatted) if not os.path.exists(main_project_dir): exit_message( f"[ERRO] Template não foi gerado. Conteúdo atual: {os.listdir(repo_root)}" ) self._update_appsettings(repo_root, args) shutil.rmtree(scaffold_dir, ignore_errors=True) semantic_released = str(args.get("semanticReleased", "true")).lower() == "true" self._commit_changes(semantic_released) print_message("Microsserviço gerado com sucesso!", Fore.GREEN) def _normalize_args(self, args): if isinstance(args, list): keys = [ "entityName", "entityPluralName", "secondaryName", "secondaryPluralName", "domain", "service", "schemaPrefix", "semanticReleased", ] if len(args) != len(keys): raise ValueError( f"Lista de argumentos inválida: esperados {len(keys)}, recebidos {len(args)}" ) args = dict(zip(keys, args)) if not isinstance(args, dict): raise ValueError(f"Formato de args inválido: {type(args)}") if not args.get("schemaPrefix"): args["schemaPrefix"] = f"{args['domain']}-{args['service']}-" return args def _update_appsettings(self, root_dir, args): appsettings_file = find_file_by_name(Path(root_dir), "appsettings.json") if not appsettings_file: print_message("[WARNING] appsettings.json não encontrado.", Fore.YELLOW) return self.update_appsettings_json( Path(appsettings_file), args["service"], args["domain"], ) def clean_dirs(self, repo_root, preserve: set): for item in os.listdir(repo_root): if item in preserve: continue path = os.path.join(repo_root, item) if os.path.isfile(path) or os.path.islink(path): os.remove(path) elif os.path.isdir(path): shutil.rmtree(path) def _commit_changes(self, semantic_released: bool = True): ci_project_dir = get_env_variable_required("CI_PROJECT_DIR") ci_repo_url = get_env_variable_required("CI_REPOSITORY_URL") exec_command("git checkout -B feature/init") ci_repo_url = re.sub(r".*@(.+)", r"git@\1", ci_repo_url) ci_repo_url = re.sub(r"/", ":", ci_repo_url, count=1) self._clean_cache_dirs() exec_command("git add -A") exec_command( f"git reset {ci_project_dir}/senior-ci/ {ci_project_dir}/venv/ senior-ci/ venv/ **/senior-ci/ **/venv/" ) exec_command("git commit -m '[CI] Primeiras fontes gerados do microsserviço.'") exec_command(f"git remote set-url --push origin '{ci_repo_url}'") branches = exec_command( "git branch -r", print_command=False, print_output=False ).output self._push_to_branches(branches or "", semantic_released) def _clean_cache_dirs(self): for path in Path.cwd().glob("**/.cache"): shutil.rmtree(path, ignore_errors=True) def _push_to_branches(self, branches: str, semantic_released: bool): self._push_branch(branches, "master") if not semantic_released: self._push_branch(branches, "develop") def _push_branch(self, branches: str, branch: str): if f"origin/{branch}" in branches: exec_command(f"git checkout {branch}") exec_command("git merge feature/init --allow-unrelated-histories") exec_command("git push") else: exec_command(f"git push --set-upstream origin feature/init:{branch}") @staticmethod def update_appsettings_json(file_path: Path, service_name: str, domain_name: str): data = load_strict_json(file_path) if "ServiceInformation" in data: data["ServiceInformation"]["Domain"] = domain_name data["ServiceInformation"]["Service"] = service_name with open(file_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=4, ensure_ascii=False)